Steer stops running subagents; draining subagents no longer trap the queue

docs/aiden-remote-api-v1.md

@@ -225,7 +225,7 @@ Selection nonces are a separate type. Workspace creation atomically revalidates
 - `GET /streams/{streamId}`
 - `GET /streams/{streamId}/events`: SSE replay via `Last-Event-ID` or `after`.
 - `POST /streams/{streamId}/cancel`: request cancellation of the authenticated parent turn stream. This is not a child/subtree control endpoint; any internal child shutdown is a Mac-owned consequence of cancelling the parent and is never separately addressable by mobile.
-- `POST /streams/{streamId}/inputs`: shared foreground admission for mid-flight input, advertised only when `/server.features` contains `chat-run-input-v1`. Requires `chat:write`, an `Idempotency-Key`, and an exact `{ "mode": "steer" | "queue", "text": string }` body bound to the displayed stream. The host derives the chat and turn from the stream record, revalidates device ownership and chat access, persists the user message durably, then admits it into the Pi runtime queue. Responses are domain receipts, not HTTP errors: `status: "admitted"` carries `queue: "steer" | "follow-up"`, `committed: true`, and `messageId`; `status: "rejected"` carries `reason: run_not_active | cancelled | capacity | invalid` plus `committed`. `committed: true` means the user message is already durable transcript history even when admission was rejected (e.g. the run finished mid-request); clients must surface it, never silently retry. The idempotency ledger replays the original outcome for a repeated request UUID—even after the stream record is evicted—and rejects conflicting payload reuse. Domain receipts replay verbatim; transport-level rejections (for example an access denial recorded mid-flight) replay as `internal_error` rather than their original code. Uncertain persistence stays `idempotency_in_flight`; control writes must not auto-retry.
+- `POST /streams/{streamId}/inputs`: shared foreground admission for mid-flight input, advertised only when `/server.features` contains `chat-run-input-v1`. Requires `chat:write`, an `Idempotency-Key`, and an exact `{ "mode": "steer" | "queue", "text": string }` body bound to the displayed stream. The host derives the chat and turn from the stream record, revalidates device ownership and chat access, persists the user message durably, then admits it into the Pi runtime queue. Responses are domain receipts, not HTTP errors: `status: "admitted"` carries `queue: "steer" | "follow-up"`, `committed: true`, and `messageId`; `status: "rejected"` carries `reason: run_not_active | cancelled | capacity | invalid` plus `committed`. `committed: true` means the user message is already durable transcript history even when admission was rejected (e.g. the run finished mid-request); clients must surface it, never silently retry. The idempotency ledger replays the original outcome for a repeated request UUID—even after the stream record is evicted—and rejects conflicting payload reuse. Domain receipts replay verbatim; transport-level rejections (for example an access denial recorded mid-flight) replay as `internal_error` rather than their original code. Uncertain persistence stays `idempotency_in_flight`; control writes must not auto-retry. An admitted `steer` also stops the run's still-running subagents so the guidance is not held behind them; the run continues and those subagents report as interrupted. `queue` never stops subagents. The wire contract is unchanged.
 - `POST /approvals/{approvalId}/respond`: `allow` or `deny` only. Allowing a
   `schedule_task` or `edit_automation` approval additionally requires the
   authenticated device's `schedule:write` grant. A device that can respond to

docs/chat-composer-busy-controls.md

@@ -20,6 +20,14 @@ All three clients show the same control while a response runs: one accent pill w
 
 Desktop Steer, Telegram `/steer`, and the Remote inputs route all enter Pi through `llmClient.admitChatRunInput`. The guidance is durable in the transcript before Pi accepts it, so a Mac restart cannot lose an accepted receipt; the run itself ends with the app, and a later delivery attempt is rejected as `run_not_active`. Desktop and Telegram do not keep a request-ID ledger: neither surface auto-retries, and Telegram does not resend edited `/steer` commands. This work is tracked by [Hermes-inspired Bot run control](plans/hermes-bot-run-control-plan.md).
 
+## Steer and running subagents
+
+Pi delivers steering only after the current tool call returns, and a `subagent` call can run for many minutes. So an accepted Steer (from any of the three surfaces) also stops that response's subagents that are still running. The response itself keeps going: each stopped subagent comes back as interrupted, with a note that the user sent new guidance, and the model reads the guidance next. A subagent that already finished keeps its result. A `subagent` call made before the guidance is delivered (for example, a second call in the same model message) launches nothing and uses none of the response's subagent budget. Queue never stops subagents. Bot conversations steer through their own runtime and are not covered yet.
+
+## Stop, failed responses and subagents
+
+Stop cancels the response's subagents and settles each one as interrupted. A subagent that ignores cancellation is cut off after a short grace period. If it is still draining after that, it keeps its model deployment quarantined, but it no longer holds the chat's turn, so queued messages and new sends are not held behind it. Deleting, copying, exporting or moving the chat still waits for it. A response that fails for any other reason also cancels any child work it still owns when it settles.
+
 ## Stop and admitted input
 
 Steer and Queue input admitted through `admitChatRunInput` (desktop Steer, Telegram `/steer`, and the phone's Steer now and Queue to run next) is saved to the transcript before Pi accepts it. Pi holds it only for the current run. Stop ends that run, and Pi does not drain its steer or follow-up queues after an aborted terminal, so:

main/services/chat-run-input-admission.test.ts

@@ -139,6 +139,38 @@ test("run input admission routes queue mode to the follow-up queue", async () =>
   assert.equal(app.steered.length, 0);
 });
 
+test("only an accepted steer stops the run's subagents, after the guidance is queued", async () => {
+  const app = fixture();
+  const events: string[] = [];
+  const queueSteer = app.generation.agent.queueSteer;
+  app.generation.agent.queueSteer = (message, options) => {
+    events.push("steer-queued");
+    return queueSteer(message, options);
+  };
+  app.generation.stopSubagentsForGuidance = () => {
+    events.push("subagents-stopped");
+  };
+
+  await app.admission.admit({ streamId: "stream-1", mode: "queue", text: "Later" });
+  assert.equal(events.length, 0);
+
+  app.setSteerReceipt({ accepted: false, reason: "capacity" });
+  const rejected = await app.admission.admit({ streamId: "stream-1", mode: "steer", text: "Full" });
+  assert.equal(rejected.admitted, false);
+  assert.deepEqual(events, ["steer-queued"]);
+
+  app.setSteerReceipt({ accepted: true, queue: "steer" });
+  events.length = 0;
+  app.generation.stopSubagentsForGuidance = () => {
+    events.push("subagents-stopped");
+    throw new Error("stop failed");
+  };
+  const admitted = await app.admission.admit({ streamId: "stream-1", mode: "steer", text: "Now" });
+  // A failed stop never turns admitted guidance into an error.
+  assert.equal(admitted.admitted, true);
+  assert.deepEqual(events, ["steer-queued", "subagents-stopped"]);
+});
+
 test("run input admission rejects unknown, foreign, and mismatched streams without consuming the draft", async () => {
   const app = fixture();
   for (const input of [

main/services/chat-run-input-admission.ts

@@ -25,6 +25,12 @@ export interface ChatRunInputGenerationRef {
   cancelRequested: boolean;
   /** Runs that take no mid-flight input (Design Studio). */
   inputClosed?: boolean;
+  /**
+   * Stop the run's still-running subagents once steering is accepted. Pi
+   * delivers steering only after the current tool call returns, so without
+   * this a steer would wait for every child to finish.
+   */
+  stopSubagentsForGuidance?(): void;
   agent: {
     queueAdmissionBlocked(): "not-active" | "cancelled" | "capacity" | undefined;
     queueSteer(message: AgentMessage, options?: PiRuntimeQueueOptions): PiRuntimeQueueReceipt;
@@ -160,6 +166,13 @@ export function createChatRunInputAdmission(deps: ChatRunInputAdmissionDeps) {
         messageId,
       };
     }
+    if (input.mode === "steer") {
+      try {
+        generation.stopSubagentsForGuidance?.();
+      } catch {
+        // The guidance is already queued; it still lands when the children end.
+      }
+    }
     return { admitted: true, queue: receipt.queue, committed: true, messageId };
   }
 

main/services/llm-client.ts

@@ -586,6 +586,8 @@ interface ActiveGeneration {
   releaseBotAuthority: () => void;
   /** Runs that take no mid-flight input (Design Studio). */
   inputClosed: boolean;
+  /** A steered send stops this run's still-running subagents. */
+  stopSubagentsForGuidance?: () => void;
 }
 
 const active = new Map<string, ActiveGeneration>();
@@ -625,14 +627,24 @@ const geminiContextCache = new GeminiContextCache({
   onWarning: (message, error) => logger.warn("pi", message, error),
 });
 
+/**
+ * Work that holds the chat's turn. A cancelled subagent that is still draining
+ * is left out: its parent already settled it, so it must not trap queued sends
+ * or new turns behind a stuck provider stream.
+ */
 function chatHasGenerationOwnership(chatId: string): boolean {
   return (
     [...initializing.values()].some((entry) => entry.chatId === chatId) ||
     [...active.values()].some((entry) => entry.chatId === chatId) ||
-    subagentRuntimeRegistry.hasChatChildren(chatId)
+    subagentRuntimeRegistry.hasLiveChatChildren(chatId)
   );
 }
 
+/** A started turn, or the chat's turn is held by live generation work. */
+function chatTurnIsHeld(chatId: string): boolean {
+  return chatTurnAdmission.isAdmitted(chatId) || chatHasGenerationOwnership(chatId);
+}
+
 function releaseGenerationSkillReservation(entry: { releaseSkillReservation: () => void }): void {
   const release = entry.releaseSkillReservation;
   entry.releaseSkillReservation = () => {};
@@ -4412,6 +4424,7 @@ export const llmClient = {
         }
       }
     };
+    subagentSupervisor?.observePendingUserGuidance(() => agent.hasUndeliveredSteer());
     const activeGeneration: ActiveGeneration = {
       agent,
       chatId: params.chatId,
@@ -4425,6 +4438,13 @@ export const llmClient = {
       computerUse,
       formFill,
       inputClosed: designWiring?.inputClosed ?? false,
+      ...(subagentSupervisor
+        ? {
+            stopSubagentsForGuidance: () => {
+              subagentSupervisor.stopLiveChildrenForGuidance();
+            },
+          }
+        : {}),
       completion: null,
       loadMonitor: initialization.loadMonitor,
       releaseSkillReservation: initialization.releaseSkillReservation,
@@ -4745,6 +4765,9 @@ export const llmClient = {
           formFill?.revoke();
           await computerUse?.close().catch(() => {});
         } finally {
+          // A run that ended for any reason (failure included) owns no more
+          // child work; nothing may keep acting for it after it settles.
+          subagentRuntimeRegistry.abortGeneration(streamId);
           unregisterAcpHost?.();
           invalidateChatContextJournal(params.chatId);
           releaseGenerationSkillReservation(activeGeneration);
@@ -4896,8 +4919,13 @@ export const llmClient = {
     return runInputAdmission.admit(input);
   },
 
+  /**
+   * Any work on the chat, including a cancelled subagent still draining (it
+   * may be inside a tool call), so deletion, copy, export and workspace moves
+   * wait for it.
+   */
   isChatBusy(chatId: string): boolean {
-    return chatTurnAdmission.isAdmitted(chatId) || chatHasGenerationOwnership(chatId);
+    return chatTurnIsHeld(chatId) || subagentRuntimeRegistry.hasChatChildren(chatId);
   },
 
   /** Whether this exact generation is still starting or running, whoever owns it. */
@@ -4917,10 +4945,14 @@ export const llmClient = {
     );
   },
 
-  /** Wait for terminal persistence without cancelling work owned by another document. */
+  /**
+   * Wait for terminal persistence without cancelling work owned by another
+   * document. Only the turn matters here, so a draining child never holds the
+   * queue or a waiting send.
+   */
   async waitForChatIdle(chatId: string): Promise<boolean> {
     const deadline = Date.now() + CHAT_CANCEL_SETTLEMENT_GRACE_MS;
-    while (this.isChatBusy(chatId)) {
+    while (chatTurnIsHeld(chatId)) {
       const remaining = deadline - Date.now();
       if (remaining <= 0) return false;
       const completions = [...active.values()]

main/services/pi-agent-runtime-harness.test.ts

@@ -2495,12 +2495,16 @@ test("managed steering is accepted only while active and queued input is durable
     message: { role: "user", content: "start", timestamp: 1 },
   });
   await atTool;
+  assert.equal(harness.hasUndeliveredSteer(), false);
   assert.deepEqual(harness.queueSteer({ role: "user", content: "new instruction", timestamp: 2 }), {
     accepted: true,
     queue: "steer",
   });
+  // Pi holds steering until the running tool returns.
+  assert.equal(harness.hasUndeliveredSteer(), true);
   releaseTool();
   assert.equal((await running).kind, "completed");
+  assert.equal(harness.hasUndeliveredSteer(), false);
   const users = (await session.buildContext()).messages.filter(
     (message) => message.role === "user",
   );

main/services/pi-agent-runtime-harness.ts

@@ -849,6 +849,8 @@ export class PiAgentRuntimeHarness {
     message: AgentMessage;
     fingerprint: string;
     visibleChatMessageId?: string;
+    /** Accepted through queueSteer: due at the next tool-batch boundary. */
+    steer?: true;
   }> = [];
   /** Visible-chat projections that app cancellation stopped waiting for. */
   private uncertainQueuedProjections: Array<{ fingerprint: string; operation: Promise<unknown> }> = [];
@@ -2470,11 +2472,20 @@ export class PiAgentRuntimeHarness {
     if (rejected) return rejected;
     const queued = snapshotQueuedMessage(message);
     if (!queued) return { accepted: false, reason: "invalid-message" };
-    this.acceptedQueuedMessages.push({ ...queued, visibleChatMessageId: options?.visibleChatMessageId });
+    this.acceptedQueuedMessages.push({
+      ...queued,
+      visibleChatMessageId: options?.visibleChatMessageId,
+      steer: true,
+    });
     this.agent.steer(queued.message);
     return { accepted: true, queue: "steer" };
   }
 
+  /** Whether accepted steering input is still waiting for Pi to deliver it. */
+  hasUndeliveredSteer(): boolean {
+    return this.acceptedQueuedMessages.some((item) => item.steer);
+  }
+
   queueFollowUp(message: AgentMessage, options?: PiRuntimeQueueOptions): PiRuntimeQueueReceipt {
     const rejected = this.rejectedQueueReceipt(message);
     if (rejected) return rejected;

main/services/subagents/agent-compatibility.test.ts

@@ -1192,10 +1192,16 @@ test("cleanup quarantine rejects new deployment work until an unresponsive child
   const running = child(registry, runtime);
   const prompt = running.prompt("Hold the local request.");
   await started.promise;
+  assert.equal(registry.hasLiveChatChildren("compatibility-chat"), true);
   running.cancel(new Error("deadline elapsed"));
   running.markCleanupPending?.();
 
   assert.equal(registry.isDeploymentQuarantined("local"), true);
+  // The undrained child still owns its deployment and workspace, but it can no
+  // longer act for the chat, so it must not keep the chat busy for new turns.
+  assert.equal(registry.hasChatChildren("compatibility-chat"), true);
+  assert.equal(registry.hasWorkspaceChildren("compatibility-workspace"), true);
+  assert.equal(registry.hasLiveChatChildren("compatibility-chat"), false);
   assert.throws(() => child(registry, runtime), /waiting for an unresponsive request to settle/u);
   assert.equal(registry.activeCount, 1);
 

main/services/subagents/child-agent-runtime.ts

@@ -451,6 +451,18 @@ export class SubagentRuntimeRegistry {
     return this.hasMatching((authority) => authority.chatId === chatId);
   }
 
+  /**
+   * Children that can still act for this chat. A cancelled child that has not
+   * drained yet is excluded: its parent already settled it as interrupted, its
+   * deployment stays quarantined until it exits, and it must not hold the chat
+   * busy (and so trap queued sends) for as long as a stuck provider stream lasts.
+   */
+  hasLiveChatChildren(chatId: string): boolean {
+    return [...this.children.values()].some(
+      (entry) => entry.authority.chatId === chatId && !entry.cancellation.signal.aborted,
+    );
+  }
+
   /** True only once a child has crossed Pi's actual provider-response boundary. */
   hasChatProviderResponse(chatId: string): boolean {
     return [...this.children.values()].some(