fix(approval): align PR77 review feedback on transport ownership

dare_framework/agent/base_agent.py

@@ -15,7 +15,12 @@
 from dare_framework.agent.status import AgentStatus
 from dare_framework.plan.types import RunResult, Task
 from dare_framework.transport.interaction.payloads import build_error_payload, build_success_payload
-from dare_framework.transport.types import EnvelopeKind, TransportEnvelope, new_envelope_id
+from dare_framework.transport.types import (
+    EnvelopeKind,
+    TransportEnvelope,
+    TransportEventType,
+    new_envelope_id,
+)
 
 if TYPE_CHECKING:
     from dare_framework.agent.builder import DareAgentBuilder, ReactAgentBuilder, SimpleChatAgentBuilder
@@ -274,6 +279,7 @@ async def _send_transport_result(
         envelope = TransportEnvelope(
             id=new_envelope_id(),
             reply_to=reply_to,
+            event_type=TransportEventType.RESULT.value,
             payload={
                 **build_success_payload(
                     kind="message",
@@ -378,6 +384,7 @@ async def _send_transport_error(
                 id=new_envelope_id(),
                 kind=EnvelopeKind.MESSAGE,
                 reply_to=envelope_id,
+                event_type=TransportEventType.ERROR.value,
                 payload=build_error_payload(
                     kind="message",
                     target=target,

dare_framework/hook/_internal/agent_event_transport_hook.py

@@ -9,7 +9,7 @@
 from dare_framework.hook.types import HookPhase
 from dare_framework.infra.component import ComponentType
 from dare_framework.transport.kernel import AgentChannel
-from dare_framework.transport.types import TransportEnvelope, new_envelope_id
+from dare_framework.transport.types import TransportEnvelope, TransportEventType, new_envelope_id
 
 _logger = logging.getLogger("dare.hook")
 
@@ -35,8 +35,8 @@ async def invoke(self, phase: HookPhase, *args: Any, **kwargs: Any) -> Any:
             payload = {}
         envelope = TransportEnvelope(
             id=new_envelope_id(),
+            event_type=TransportEventType.HOOK.value,
             payload={
-                "type": "hook",
                 "phase": phase.value,
                 "payload": payload,
             },

dare_framework/tool/_internal/control/approval_manager.py

@@ -147,6 +147,10 @@ class ApprovalEvaluation:
 class _PendingApproval:
     request: PendingApprovalRequest
     fingerprint: str
+    # Track all sessions that are currently blocked on this deduplicated request.
+    # The first requester is also recorded so session-filtered polling can match it
+    # even after subsequent evaluate() calls deduplicate to the same request id.
+    session_ids: set[str] = field(default_factory=set)
     event: asyncio.Event = field(default_factory=asyncio.Event)
     resolution: ApprovalDecision | None = None
 
@@ -217,8 +221,10 @@ def __init__(
         self._pending_by_id: dict[str, _PendingApproval] = {}
         self._pending_by_fingerprint: dict[str, _PendingApproval] = {}
         self._resolved_by_id: dict[str, ApprovalDecision] = {}
-        self._pending_available = asyncio.Event()
         self._lock = asyncio.Lock()
+        # Condition-based wakeups avoid tight loops when polling with a session filter
+        # while unrelated pending requests exist.
+        self._pending_state_changed = asyncio.Condition(self._lock)
 
     @classmethod
     def from_paths(cls, *, workspace_dir: str | Path, user_dir: str | Path) -> ToolApprovalManager:
@@ -231,31 +237,36 @@ def list_pending(self) -> list[PendingApprovalRequest]:
         pending.sort(key=lambda item: (item.created_at, item.request_id))
         return pending
 
-    async def poll_pending(self, *, timeout_seconds: float | None = None) -> PendingApprovalRequest | None:
-        """Return the oldest pending approval request, optionally waiting for one."""
+    async def poll_pending(
+        self,
+        *,
+        timeout_seconds: float | None = None,
+        session_id: str | None = None,
+    ) -> PendingApprovalRequest | None:
+        """Return the oldest pending approval request, optionally filtered by session."""
         if timeout_seconds is not None and timeout_seconds < 0:
             raise ValueError("timeout_seconds must be >= 0")
 
         loop = asyncio.get_running_loop()
         deadline = None if timeout_seconds is None else loop.time() + timeout_seconds
-        while True:
-            async with self._lock:
-                request = self._oldest_pending_locked()
+
+        async with self._pending_state_changed:
+            while True:
+                request = self._oldest_pending_locked(session_id=session_id)
                 if request is not None:
                     return request
-                wait_event = self._pending_available
 
-            if deadline is None:
-                await wait_event.wait()
-                continue
+                if deadline is None:
+                    await self._pending_state_changed.wait()
+                    continue
 
-            remaining = deadline - loop.time()
-            if remaining <= 0:
-                return None
-            try:
-                await asyncio.wait_for(wait_event.wait(), timeout=remaining)
-            except asyncio.TimeoutError:
-                return None
+                remaining = deadline - loop.time()
+                if remaining <= 0:
+                    return None
+                try:
+                    await asyncio.wait_for(self._pending_state_changed.wait(), timeout=remaining)
+                except asyncio.TimeoutError:
+                    return None
 
     def list_rules(self) -> list[ApprovalRule]:
         combined = [
@@ -278,7 +289,8 @@ async def evaluate(
         command = _extract_command(params)
         fingerprint = _request_fingerprint(capability_id, params_hash)
 
-        async with self._lock:
+        # Use the condition lock consistently for pending-state mutations.
+        async with self._pending_state_changed:
             matched_rule = self._find_matching_rule(
                 capability_id=capability_id,
                 params_hash=params_hash,
@@ -313,9 +325,16 @@ async def evaluate(
                     created_at=self._time_fn(),
                 )
                 existing = _PendingApproval(request=request, fingerprint=fingerprint)
+                self._track_pending_session_locked(existing, session_id)
                 self._pending_by_fingerprint[fingerprint] = existing
                 self._pending_by_id[request.request_id] = existing
-                self._pending_available.set()
+                self._pending_state_changed.notify_all()
+            else:
+                # A deduplicated request can gain new interested sessions later.
+                # Wake session-filtered poll waiters when that subscriber set expands.
+                added_session = self._track_pending_session_locked(existing, session_id)
+                if added_session:
+                    self._pending_state_changed.notify_all()
 
             return ApprovalEvaluation(
                 status=ApprovalEvaluationStatus.PENDING,
@@ -329,7 +348,9 @@ async def wait_for_resolution(
         *,
         timeout_seconds: float | None = None,
     ) -> ApprovalDecision:
-        async with self._lock:
+        # Keep all pending/resolution map access on the same condition-backed lock
+        # so concurrency audits only need to reason about one synchronization surface.
+        async with self._pending_state_changed:
             resolved = self._resolved_by_id.pop(request_id, None)
             if resolved is not None:
                 return resolved
@@ -345,7 +366,7 @@ async def wait_for_resolution(
 
         if pending.resolution is None:
             raise RuntimeError(f"Approval request resolved without decision: {request_id}")
-        async with self._lock:
+        async with self._pending_state_changed:
             self._resolved_by_id.pop(request_id, None)
         return pending.resolution
 
@@ -382,7 +403,7 @@ async def deny(
         )
 
     async def revoke(self, rule_id: str) -> bool:
-        async with self._lock:
+        async with self._pending_state_changed:
             removed = self._remove_rule(rule_id)
             if removed:
                 self._persist_rules_for_scope(removed.scope)
@@ -397,7 +418,8 @@ async def _resolve_request(
         matcher: ApprovalMatcherKind,
         matcher_value: str | None,
     ) -> ApprovalRule | None:
-        async with self._lock:
+        # Keep pending-state transitions on the condition lock to avoid mixed styles.
+        async with self._pending_state_changed:
             pending = self._pending_by_id.get(request_id)
             if pending is None:
                 raise KeyError(f"Unknown approval request: {request_id}")
@@ -419,8 +441,7 @@ async def _resolve_request(
             self._resolved_by_id[request_id] = decision
             self._pending_by_id.pop(request_id, None)
             self._pending_by_fingerprint.pop(pending.fingerprint, None)
-            if not self._pending_by_id:
-                self._pending_available.clear()
+            self._pending_state_changed.notify_all()
             return rule
 
     def _append_rule(self, rule: ApprovalRule) -> None:
@@ -509,15 +530,33 @@ def _find_matching_rule(
                 return rule
         return None
 
-    def _oldest_pending_locked(self) -> PendingApprovalRequest | None:
+    def _oldest_pending_locked(self, *, session_id: str | None = None) -> PendingApprovalRequest | None:
         if not self._pending_by_id:
             return None
+        candidates = list(self._pending_by_id.values())
+        if session_id is not None:
+            candidates = [
+                item
+                for item in candidates
+                if session_id in item.session_ids or item.request.session_id == session_id
+            ]
+            if not candidates:
+                return None
         oldest = min(
-            self._pending_by_id.values(),
+            candidates,
             key=lambda item: (item.request.created_at, item.request.request_id),
         )
         return oldest.request
 
+    @staticmethod
+    def _track_pending_session_locked(pending: _PendingApproval, session_id: str | None) -> bool:
+        if isinstance(session_id, str) and session_id:
+            if session_id in pending.session_ids:
+                return False
+            pending.session_ids.add(session_id)
+            return True
+        return False
+
 
 def _rule_matches(
     rule: ApprovalRule,