feat(console): Fleet Room composer — @-address a member

a2a_impl/executor.py

@@ -117,10 +117,12 @@ def _notify_terminal(outcome: TurnOutcome) -> None:
 
 
 # A progress hook (ADR 0051) the host can register to observe a turn's realtime
-# frames — fired at turn start (``turn_started``, which carries the task_id) and on
-# each tool start/end with ``(context_id, task_id, frame)``. No-op when unset, so live
-# turns (which already stream over their own SSE) pay nothing; the background-subagents
-# publisher uses it to surface a detached turn's progress on the event bus.
+# frames — fired at turn start (``turn_started``, which carries the task_id + a
+# ``resumed`` flag for HITL continuations), on each tool start/end, and on a HITL
+# pause (``input_required``, with the human-readable prompt) with
+# ``(context_id, task_id, frame)``. No-op when unset, so live turns (which already
+# stream over their own SSE) pay nothing; the host publisher uses it to surface
+# detached-turn progress and HITL pauses on the event bus.
 _ON_PROGRESS: list[Callable[[str, str, dict], None] | None] = [None]
 
 
@@ -242,8 +244,12 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non
             )
         await updater.start_work()
         # Realtime: announce the turn (carries the task_id — the handle a host needs to
-        # control or re-attach to this turn) before any work runs (ADR 0051).
-        _notify_progress(context.context_id, context.task_id, {"phase": "turn_started"})
+        # control or re-attach to this turn) before any work runs (ADR 0051). `resumed`
+        # marks a HITL continuation (the parked task got its answer), so a host can flip
+        # a "needs approval" surface back to "running" (#2132).
+        _notify_progress(
+            context.context_id, context.task_id, {"phase": "turn_started", "resumed": resume}
+        )
 
         text = context.get_user_input()
         images = _extract_image_parts(context)
@@ -508,6 +514,15 @@ def _outcome(state: str, final_text: str) -> TurnOutcome:
 
                 elif event_type == "input_required":
                     await _flush_text()  # persist any answer text streamed before the pause
+                    # Surface the pause to the host hook (ADR 0051) so it can publish a
+                    # "needs approval" bus event — a fleet console can then show a pending
+                    # HITL without polling. Best-effort by contract (_notify_progress
+                    # swallows everything), so it can never break the pause itself.
+                    _notify_progress(
+                        context.context_id,
+                        context.task_id,
+                        {"phase": "input_required", "prompt": _hitl_prompt(payload)},
+                    )
                     # Human-readable prompt for plain consumers; the full
                     # form/approval payload rides a protoAgent-local hitl-v1
                     # DataPart so the console renders the form / approval card.

apps/web/e2e/fleet.spec.ts

@@ -219,60 +219,67 @@ test("discover → add to fleet → switch into the remote member (ADR 0042 §I)
   await expect(page.locator(".fleet-row", { hasText: "remy" })).toHaveCount(0);
 });
 
-// ── Command-palette fleet toggle (#1769) ───────────────────────────────────────────
-// Toggle a local, non-host member on/off straight from ⌘K — no Settings dive. The
-// picker lists togglable members with their live process state; picking one flips it and
-// toasts. Reuses this file's per-spec fleet scope (beforeEach resets to baseline).
-
-// The DS palette morphs views with a popLayout cross-fade, so the exiting root list lingers
-// in the DOM for a beat next to the entering sub-view. A sub-view row's accessible name is
-// "<name> <state>" (e.g. "ava on") — and the sub-view states ("on"/"off") never collide with
-// the root quick-chat's hints ("switch"/"stopped"/"unreachable"), so keying every assertion
-// on the exact "<name> <state>" name sidesteps the transient overlap entirely.
-async function openToggleFleet(page) {
+// ── Folded-in fleet controls (#1733 quick-chat + #1769 toggle → the Fleet Room) ─────
+// The per-member root commands and the `Toggle Fleet Agent ▸` submorph are gone: member
+// names ride the Fleet Room command's keywords, and the roster row carries DM / open /
+// start / stop. These tests pin the fold — the old flows keep working, one hop away.
+
+test("⌘K root: member names surface the Fleet Room; the old per-member commands are gone", async ({ page }) => {
+  await page.goto("/app/", { waitUntil: "load" });
   await page.keyboard.press("ControlOrMeta+k");
   await expect(page.locator(".pl-cmdk__panel")).toBeVisible();
-  // A reopen momentarily re-morphs the last sub-view back to the root view (the DS palette
-  // resets its stack on open), so the sub-view's search box lingers next to the root one for a
-  // beat — wait for it to leave before filling, or `.pl-cmdk-commands__input` is ambiguous.
-  await expect(page.getByPlaceholder("Toggle a fleet agent on/off…")).toHaveCount(0);
-  // Filter the root list to the toggle command, then enter its submorph.
-  await page.locator(".pl-cmdk__panel .pl-cmdk-commands__input").fill("Toggle Fleet Agent");
-  await page.getByRole("option", { name: "Toggle Fleet Agent" }).click();
-  await expect(page.locator(".pl-cmdk__title")).toHaveText("Toggle Fleet Agent"); // in the sub-view
-}
-
-const row = (page, name, state) => page.getByRole("option", { name: `${name} ${state}`, exact: true });
+  const input = page.locator(".pl-cmdk__panel .pl-cmdk-commands__input");
+  // The toggle submorph is folded away.
+  await input.fill("Toggle Fleet Agent");
+  await expect(page.getByRole("option", { name: "Toggle Fleet Agent" })).toHaveCount(0);
+  // Typing a member's name routes to the room (roster keywords), not a per-member row.
+  await input.fill("ava");
+  await expect(page.getByRole("option", { name: "Fleet Room" })).toBeVisible();
+  await expect(page.getByRole("option", { name: /^ava\b/ })).toHaveCount(0);
+  await page.getByRole("option", { name: "Fleet Room" }).click();
+  await expect(page.locator(".flr")).toBeVisible();
+});
 
-test("⌘K → Toggle Fleet Agent lists non-host members with state and flips one", async ({ page }) => {
+test("Fleet Room roster: stop a running member, start a stopped one (folded #1769)", async ({ page }) => {
   await page.goto("/app/", { waitUntil: "load" });
-  await openToggleFleet(page);
-
-  // Local members are listed with their live process state; the host (main) never is.
-  await expect(row(page, "ava", "on")).toBeVisible(); // running in baseline
-  await expect(row(page, "roxy", "off")).toBeVisible(); // stopped, still listed
-  await expect(page.getByRole("option", { name: /^main\b/ })).toHaveCount(0); // host excluded
+  await openFleetRoom(page);
+  const room = page.locator(".flr");
 
-  // Toggle ava off — the palette closes, a toast confirms, and the roster flips.
-  await row(page, "ava", "on").click();
-  await expect(page.locator(".pl-cmdk__panel")).toHaveCount(0);
+  // Stop ava (running in baseline) straight from her roster row; the dot flips on the
+  // invalidated poll. The host (main) never gets a toggle — it serves this console.
+  await expect(room.locator(".flr__member", { hasText: "main" }).getByRole("button", { name: /^(Stop|Start) main$/ })).toHaveCount(0);
+  await room.locator(".flr__member", { hasText: "ava" }).getByRole("button", { name: "Stop ava" }).click();
   await expect(page.locator(".pl-toast", { hasText: "Stopping ava" })).toBeVisible();
+  await expect(room.locator(".flr__member", { hasText: "ava" }).locator(".flr__dot--stopped")).toBeVisible();
 
-  // Reopen the picker: ava now reads "off" (its running state flipped on the invalidated poll).
-  await openToggleFleet(page);
-  await expect(row(page, "ava", "off")).toBeVisible();
+  // Start roxy (stopped in baseline).
+  await room.locator(".flr__member", { hasText: "roxy" }).getByRole("button", { name: "Start roxy" }).click();
+  await expect(page.locator(".pl-toast", { hasText: "Starting roxy" })).toBeVisible();
+  await expect(room.locator(".flr__member", { hasText: "roxy" }).locator(".flr__dot--online")).toBeVisible();
 });
 
-test("⌘K → Toggle Fleet Agent starts a stopped member", async ({ page }) => {
+test("Fleet Room: a parked member turn shows 'needs approval', then hands back (#2132)", async ({ page }, testInfo) => {
+  // setExtraHTTPHeaders REPLACES the set — re-carry the fleet scope alongside the HITL gate.
+  await page.setExtraHTTPHeaders({
+    "x-e2e-fleet": `fleet-spec-${testInfo.parallelIndex}`,
+    "x-e2e-hitl": "ava",
+  });
   await page.goto("/app/", { waitUntil: "load" });
-  await openToggleFleet(page);
+  await openFleetRoom(page);
+  const room = page.locator(".flr");
+  const ava = room.locator(".flr__member", { hasText: "ava" });
 
-  await expect(row(page, "roxy", "off")).toBeVisible(); // stopped in baseline
-  await row(page, "roxy", "off").click();
-  await expect(page.locator(".pl-toast", { hasText: "Starting roxy" })).toBeVisible();
+  // ava's stream emits turn.input_required → attention pill + the actionable feed row.
+  await expect(ava.locator(".flr__pill--attn")).toBeVisible();
+  await expect(room.locator(".flr-feed__event", { hasText: "needs your approval" }).first()).toBeVisible();
+
+  // The answer lands (turn.resumed) — needs-approval hands back to a live "running" pill…
+  await expect(ava.locator(".flr__pill--run")).toBeVisible();
+  await expect(room.locator(".flr-feed__event", { hasText: "resumed — input received" }).first()).toBeVisible();
 
-  await openToggleFleet(page);
-  await expect(row(page, "roxy", "on")).toBeVisible();
+  // …and the terminal turn.usage clears it.
+  await expect(ava.locator(".flr__pill--run")).toHaveCount(0, { timeout: 6000 });
+  await expect(ava.locator(".flr__pill--attn")).toHaveCount(0);
 });
 
 async function openFleetRoom(page) {
@@ -298,6 +305,8 @@ test("⌘K → Fleet Room: presence, DM a member (the wired chat), broadcast", a
   // (placeholder names them). Back returns to the roster.
   await room.locator(".flr__member", { hasText: "ava" }).locator(".flr__who").click();
   await expect(page.getByPlaceholder(/Message ava/i)).toBeVisible();
+  // The DM header names the member (DmTitle store) — not the old generic "Direct message".
+  await expect(page.locator(".pl-cmdk__title")).toHaveText("@ava");
   await page.locator(".pl-cmdk__back").click();
   await expect(room.locator(".flr__composer")).toBeVisible();
 
@@ -319,3 +328,29 @@ test("⌘K → Fleet Room shows the roster + live activity feed side by side", a
   // pushes activity/inbox/goal frames, so a mapped event lands in the column.
   await expect(room.locator(".flr-feed__event").first()).toBeVisible({ timeout: 6000 });
 });
+
+test("⌘K → Fleet Room: @-address a member in the composer, then send opens its DM", async ({ page }) => {
+  await page.goto("/app/", { waitUntil: "load" });
+  await openFleetRoom(page);
+  const room = page.locator(".flr");
+  // Typing "@" opens a member picker; picking sets the address chip.
+  await room.locator(".flr__input").fill("@ava");
+  await room.locator(".flr__mention", { hasText: "ava" }).click();
+  await expect(room.locator(".flr__target")).toContainText("@ava");
+  // Type a message and send → morphs into ava's DM (the wired chat), message pre-sent.
+  await room.locator(".flr__input").fill("ship it");
+  await room.locator(".flr__send").click();
+  await expect(page.getByPlaceholder(/Message ava/i)).toBeVisible();
+});
+
+test("⌘K → Fleet Room: a TYPED @name addresses that member without using the picker", async ({ page }) => {
+  await page.goto("/app/", { waitUntil: "load" });
+  await openFleetRoom(page);
+  const room = page.locator(".flr");
+  // Never touch the picker — just type "@ava <message>" and send, the way people actually
+  // type. It must address ava (open its DM), NOT broadcast the literal text.
+  await room.locator(".flr__input").fill("@ava ship it");
+  await room.locator(".flr__send").click();
+  await expect(page.getByPlaceholder(/Message ava/i)).toBeVisible();
+  await expect(page.locator(".pl-toast", { hasText: /Broadcast to/ })).toHaveCount(0);
+});

apps/web/e2e/mock-server.mjs

@@ -537,7 +537,18 @@ const server = createServer(async (req, res) => {
       ? [setTimeout(() => frame("turn.started", { session_id: turnSession, origin: "scheduler" }), 300),
          setTimeout(() => frame("turn.finished", { session_id: turnSession }), 2500)]
       : [];
-    req.on("close", () => { clearInterval(t); goals.forEach(clearTimeout); turns.forEach(clearTimeout); });
+    // Fleet Room HITL lifecycle (#2132), slug-gated: the spec names the member whose
+    // proxied stream should park a turn (x-e2e-hitl: <slug>) and ONLY that stream emits
+    // the sequence — pause → answer lands → terminal — so the roster pill walks
+    // needs-approval → running → cleared, deterministically per connection.
+    const memberSlug = (url.pathname.match(/^\/agents\/([^/]+)\//) || [])[1] || "";
+    const hitlSlug = req.headers["x-e2e-hitl"];
+    const hitl = hitlSlug && memberSlug === hitlSlug
+      ? [setTimeout(() => frame("turn.input_required", { task_id: "t-hitl", context_id: "sess-hitl", prompt: "Approve the deploy?" }), 400),
+         setTimeout(() => frame("turn.resumed", { task_id: "t-hitl", context_id: "sess-hitl" }), 1800),
+         setTimeout(() => frame("turn.usage", { task_id: "t-hitl", context_id: "sess-hitl", state: "completed", input_tokens: 10, output_tokens: 5 }), 3200)]
+      : [];
+    req.on("close", () => { clearInterval(t); goals.forEach(clearTimeout); turns.forEach(clearTimeout); hitl.forEach(clearTimeout); });
     return;
   }
   if (pathname.startsWith("/api/")) {