fix(delegation): runtime handles 202+queued; canvas surfaces delegation rows

canvas/src/components/tabs/chat/AgentCommsPanel.tsx

@@ -59,6 +59,34 @@ function resolveName(id: string): string {
 }
 
 export function toCommMessage(entry: ActivityEntry, workspaceId: string): CommMessage | null {
+  // delegation activity rows are written by the platform's /delegate
+  // handler. They're always outbound from this workspace's POV (the
+  // platform proxies the A2A on our behalf). Two methods:
+  //   - "delegate"        — the initial outbound; status pending/dispatched
+  //   - "delegate_result" — the eventual reply; status completed/queued/failed
+  // We surface them in Agent Comms because they ARE agent-to-agent
+  // calls; without this branch they'd be dropped by the activity_type
+  // filter and the user would see "No agent-to-agent communications yet"
+  // even when the director made delegations.
+  if (entry.activity_type === "delegation") {
+    const peerId = entry.target_id || "";
+    if (!peerId) return null;
+    return {
+      id: entry.id,
+      flow: "out",
+      peerName: resolveName(peerId),
+      peerId,
+      // Prefer summary (set by the platform with a human-readable
+      // string like "Delegating to X" or "Delegation queued — target
+      // at capacity"). Fall back to request body for older rows that
+      // pre-date the summary column being populated.
+      text: entry.summary || extractRequestText(entry.request_body) || "(delegation)",
+      responseText: entry.response_body ? extractResponseText(entry.response_body) : null,
+      status: entry.status || "ok",
+      timestamp: entry.created_at,
+    };
+  }
+
   // a2a_receive activity rows come in two shapes:
   //
   //   1. Real incoming call (a peer called us): source_id = the peer,
@@ -132,7 +160,11 @@ export function AgentCommsPanel({ workspaceId }: { workspaceId: string }) {
     api.get<ActivityEntry[]>(`/workspaces/${workspaceId}/activity?source=agent&limit=50`)
       .then((entries) => {
         const filtered = (entries ?? [])
-          .filter((e) => e.activity_type === "a2a_send" || e.activity_type === "a2a_receive")
+          .filter((e) =>
+            e.activity_type === "a2a_send" ||
+            e.activity_type === "a2a_receive" ||
+            e.activity_type === "delegation",
+          )
           .reverse();
         const msgs: CommMessage[] = [];
         for (const e of filtered) {
@@ -181,14 +213,34 @@ export function AgentCommsPanel({ workspaceId }: { workspaceId: string }) {
     ws.onmessage = (event) => {
       try {
         const msg = JSON.parse(event.data);
-        if (msg.event === "ACTIVITY_LOGGED" && msg.workspace_id === workspaceId) {
+        if (msg.workspace_id !== workspaceId) return;
+
+        // Two live-update paths:
+        //   1. ACTIVITY_LOGGED — fired by the LogActivity helper for
+        //      a2a_send / a2a_receive (and delegation rows IF the
+        //      delegation handler is ever refactored to use it). Today
+        //      the platform's delegation handlers do direct INSERT
+        //      INTO activity_logs WITHOUT firing ACTIVITY_LOGGED, so
+        //      the delegation branch here is defensive — it'd light
+        //      up automatically the day delegation handlers are
+        //      refactored to call LogActivity.
+        //   2. DELEGATION_SENT / DELEGATION_STATUS / DELEGATION_COMPLETE
+        //      / DELEGATION_FAILED — fired by the platform's delegation
+        //      handlers directly. These are the ONLY live signals the
+        //      panel currently has for delegation rows; the GET on
+        //      mount (which reads from activity_logs) shows them, but
+        //      without this branch, nothing surfaced live until the
+        //      next remount. Synthesise an ActivityEntry from the
+        //      payload so toCommMessage's existing delegation branch
+        //      handles them identically to the GET path.
+        let entry: ActivityEntry | null = null;
+        if (msg.event === "ACTIVITY_LOGGED") {
           const p = msg.payload || {};
           const type = p.activity_type as string;
           const sourceId = p.source_id as string | null;
           if (!sourceId) return; // canvas-initiated, not agent comms
-          if (type !== "a2a_send" && type !== "a2a_receive") return;
-
-          const entry: ActivityEntry = {
+          if (type !== "a2a_send" && type !== "a2a_receive" && type !== "delegation") return;
+          entry = {
             id: p.id as string || crypto.randomUUID(),
             activity_type: type,
             source_id: sourceId,
@@ -200,13 +252,56 @@ export function AgentCommsPanel({ workspaceId }: { workspaceId: string }) {
             status: p.status as string || "ok",
             created_at: msg.timestamp || new Date().toISOString(),
           };
-          const m = toCommMessage(entry, workspaceId);
-          if (m) {
-            const key = `${m.timestamp}:${m.flow}:${m.peerId}`;
-            if (seenKeys.current.has(key)) return;
-            seenKeys.current.add(key);
-            setMessages((prev) => [...prev, m]);
+        } else if (
+          msg.event === "DELEGATION_SENT" ||
+          msg.event === "DELEGATION_STATUS" ||
+          msg.event === "DELEGATION_COMPLETE" ||
+          msg.event === "DELEGATION_FAILED"
+        ) {
+          const p = msg.payload || {};
+          const targetId = (p.target_id as string) || "";
+          if (!targetId) return;
+          // Map event → status. DELEGATION_STATUS payload includes its
+          // own `status` field (queued / dispatched). Other events have
+          // implicit status: SENT → pending, COMPLETE → completed,
+          // FAILED → failed.
+          let status: string;
+          let summary: string;
+          if (msg.event === "DELEGATION_STATUS") {
+            status = (p.status as string) || "queued";
+            summary = `Delegation ${status}`;
+          } else if (msg.event === "DELEGATION_COMPLETE") {
+            status = "completed";
+            summary = `Delegation completed (${(p.response_preview as string)?.slice(0, 60) || ""})`;
+          } else if (msg.event === "DELEGATION_FAILED") {
+            status = "failed";
+            summary = `Delegation failed: ${(p.error as string) || "unknown"}`;
+          } else {
+            status = "pending";
+            summary = `Delegating to ${(p.target_id as string)?.slice(0, 8) || "peer"}`;
           }
+          entry = {
+            id: (p.delegation_id as string) || crypto.randomUUID(),
+            activity_type: "delegation",
+            source_id: workspaceId,
+            target_id: targetId,
+            method: msg.event === "DELEGATION_SENT" ? "delegate" : "delegate_result",
+            summary,
+            request_body: null,
+            response_body: null,
+            status,
+            created_at: msg.timestamp || new Date().toISOString(),
+          };
+        } else {
+          return;
+        }
+
+        const m = toCommMessage(entry, workspaceId);
+        if (m) {
+          const key = `${m.timestamp}:${m.flow}:${m.peerId}`;
+          if (seenKeys.current.has(key)) return;
+          seenKeys.current.add(key);
+          setMessages((prev) => [...prev, m]);
         }
       } catch { /* ignore */ }
     };

canvas/src/components/tabs/chat/__tests__/AgentCommsPanel.test.ts

@@ -110,4 +110,65 @@ describe("toCommMessage — flow derivation", () => {
     );
     expect(m!.status).toBe("error");
   });
+
+  // --- delegation rows ---
+  // The platform's /delegate handler writes activity_type='delegation'
+  // for both the initial outbound (method='delegate') and the eventual
+  // reply (method='delegate_result', status=queued|completed|failed).
+  // Pre-fix the panel filtered these out and showed "no agent comms"
+  // even when 6+ delegations existed in the DB.
+
+  it("delegation 'delegate' row maps as outbound to target", () => {
+    const m = toCommMessage(
+      makeEntry({
+        activity_type: "delegation",
+        method: "delegate",
+        source_id: SELF,
+        target_id: PEER,
+        summary: "Delegating to ws-peer",
+        status: "pending",
+      }),
+      SELF,
+    );
+    expect(m).toBeTruthy();
+    expect(m!.flow).toBe("out");
+    expect(m!.peerId).toBe(PEER);
+    expect(m!.peerName).toBe("Peer Agent");
+    expect(m!.text).toBe("Delegating to ws-peer");
+    expect(m!.status).toBe("pending");
+  });
+
+  it("delegation 'delegate_result' queued row preserves status='queued'", () => {
+    // The "queued" status is the load-bearing signal the LLM uses to
+    // decide whether to wait or fall back. If toCommMessage drops or
+    // rewrites it, the UI loses the ability to show the "peer busy,
+    // will reply" affordance.
+    const m = toCommMessage(
+      makeEntry({
+        activity_type: "delegation",
+        method: "delegate_result",
+        source_id: SELF,
+        target_id: PEER,
+        summary: "Delegation queued — target at capacity",
+        status: "queued",
+      }),
+      SELF,
+    );
+    expect(m!.status).toBe("queued");
+    expect(m!.text).toContain("queued");
+  });
+
+  it("delegation row with no target_id returns null", () => {
+    // Defensive: a delegation row missing target_id can't be rendered
+    // (we wouldn't know which peer to attribute it to). Drop instead
+    // of rendering a ghost.
+    const m = toCommMessage(
+      makeEntry({
+        activity_type: "delegation",
+        target_id: null,
+      }),
+      SELF,
+    );
+    expect(m).toBeNull();
+  });
 });

workspace/builtin_tools/delegation.py

@@ -39,6 +39,13 @@
 class DelegationStatus(str, Enum):
     PENDING = "pending"
     IN_PROGRESS = "in_progress"
+    # QUEUED: peer's a2a-proxy returned HTTP 202 + {queued: true}, meaning
+    # the peer is mid-task and the request was placed in a drain queue.
+    # The reply will arrive via the platform's stitch path when the
+    # peer finishes its current work. The LLM should WAIT, not retry,
+    # and definitely not fall back to doing the work itself — see the
+    # check_delegation_status docstring for the prompt-side guidance.
+    QUEUED = "queued"
     COMPLETED = "completed"
     FAILED = "failed"
 
@@ -119,6 +126,69 @@ async def _record_delegation_on_platform(task_id: str, target_workspace_id: str,
         logger.debug("Delegation record failed (best-effort): %s", e)
 
 
+async def _refresh_queued_from_platform(task_id: str) -> bool:
+    """Lazy-refresh a QUEUED delegation's local state from the platform.
+
+    Called by check_delegation_status when local status is QUEUED. The
+    platform's drain stitch (a2a_queue.go) updates the delegate_result
+    activity_logs row when a queued delegation eventually completes,
+    but it has no callback to this runtime — without this lazy refresh,
+    the LLM polling check_delegation_status would see "queued" forever
+    even after the platform has the result.
+
+    Returns True if the local delegation was updated to a terminal state
+    (completed/failed), False otherwise. Best-effort — network/parse
+    errors leave the local state untouched and let the next call retry.
+    """
+    delegation = _delegations.get(task_id)
+    if not delegation:
+        return False
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            resp = await client.get(
+                f"{PLATFORM_URL}/workspaces/{WORKSPACE_ID}/delegations",
+                headers={},
+            )
+            if resp.status_code != 200:
+                return False
+            entries = resp.json()
+            if not isinstance(entries, list):
+                return False
+    except Exception as e:
+        logger.debug("refresh queued delegation %s: %s", task_id, e)
+        return False
+    # Find the latest delegate_result row matching our task_id.
+    # Platform list is newest-first; the first match is the freshest.
+    for entry in entries:
+        if entry.get("delegation_id") != task_id:
+            continue
+        if entry.get("type") != "delegation":
+            continue
+        # Only delegate_result rows carry the eventual outcome; the
+        # initial 'delegate' row stays at status='pending' even after
+        # the result lands. Filtering on summary text is brittle, but
+        # the rows from the LIST endpoint don't include `method`. The
+        # `delegate_result` rows are the ones with `error` (failure)
+        # or `response_preview` (success) populated — pick those.
+        status = entry.get("status", "")
+        if status == "completed":
+            delegation.status = DelegationStatus.COMPLETED
+            delegation.result = entry.get("response_preview", "")
+            await _notify_completion(task_id, delegation.workspace_id, "completed")
+            return True
+        if status == "failed":
+            delegation.status = DelegationStatus.FAILED
+            delegation.error = entry.get("error", "")
+            await _notify_completion(task_id, delegation.workspace_id, "failed")
+            return True
+        # status == "queued" / "pending" / "dispatched": platform hasn't
+        # resolved yet; leave local state unchanged so the next poll
+        # retries. Don't break — keep scanning in case there's a newer
+        # entry for the same task_id (possible if the same delegation
+        # was retried).
+    return False
+
+
 async def _update_delegation_on_platform(task_id: str, status: str, error: str = "", response_preview: str = ""):
     """Mirror status changes to the platform's activity_logs (#64 fix).
 
@@ -212,6 +282,39 @@ async def _execute_delegation(task_id: str, workspace_id: str, task: str):
                         },
                     )
 
+                    # HTTP 202 + {queued: true} = peer's a2a-proxy
+                    # accepted the request but the peer's runtime is
+                    # mid-task. Platform-side drain will deliver the
+                    # reply asynchronously. Mark QUEUED locally so
+                    # check_delegation_status can surface that state
+                    # to the LLM with explicit "wait, don't bypass"
+                    # guidance. Do NOT mark FAILED — the request is
+                    # alive in the platform's queue, not lost.
+                    #
+                    # Without this branch, the loop falls through, the
+                    # `if "error" in result` line below references an
+                    # unbound `result`, and the eventual FAILED status
+                    # leads the LLM to conclude the peer is permanently
+                    # unavailable — at which point it does the delegated
+                    # work itself, defeating the whole orchestration.
+                    if a2a_resp.status_code == 202:
+                        try:
+                            queued_body = a2a_resp.json()
+                        except Exception:
+                            queued_body = {}
+                        if queued_body.get("queued") is True:
+                            delegation.status = DelegationStatus.QUEUED
+                            log_event(
+                                event_type="delegation", action="delegate",
+                                resource=workspace_id, outcome="queued",
+                                trace_id=task_id, attempt=attempt + 1,
+                            )
+                            await _notify_completion(task_id, workspace_id, "queued")
+                            await _update_delegation_on_platform(
+                                task_id, "queued", "", "",
+                            )
+                            return
+
                     if a2a_resp.status_code == 200:
                         try:
                             result = a2a_resp.json()
@@ -324,6 +427,23 @@ async def check_delegation_status(
 ) -> dict:
     """Check the status of a delegated task, or list all active delegations.
 
+    Status semantics — IMPORTANT:
+
+    - "pending" / "in_progress" → peer is actively working. Wait and check again.
+    - "queued" → peer's a2a-proxy accepted the call but the peer is
+      processing a prior task. The reply WILL arrive — the platform's
+      drain re-dispatches when the peer is free. This tool transparently
+      polls the platform for the eventual outcome on each call, so
+      keep polling check_delegation_status periodically and you'll see
+      the status flip to "completed" / "failed" automatically.
+      Do NOT retry the delegation. Do NOT do the work yourself.
+      Acknowledge to the user that the peer is busy and will reply,
+      then continue with other delegations or check back later.
+    - "completed" → result is in the `result` field.
+    - "failed" → real failure (network, peer crashed, etc.). The
+      `error` field has the cause. Only fall back to doing the work
+      yourself if status is "failed", never if status is "queued".
+
     Args:
         task_id: The task_id returned by delegate_to_workspace. If empty, lists all delegations.
 
@@ -351,6 +471,16 @@ async def check_delegation_status(
     if not delegation:
         return {"error": f"No delegation found with task_id {task_id}"}
 
+    # Lazy refresh for QUEUED entries: the platform's drain stitch
+    # updates its activity_logs row when the queued delegation
+    # eventually completes, but doesn't push back to this runtime.
+    # Without this refresh, the LLM polling here would see "queued"
+    # forever even after the result is available — exactly the bug
+    # the upstream director-bypass docstring guidance warned against.
+    if delegation.status == DelegationStatus.QUEUED:
+        await _refresh_queued_from_platform(task_id)
+        # delegation is the same dict entry — _refresh mutates in-place.
+
     result = {
         "task_id": task_id,
         "workspace_id": delegation.workspace_id,