From 6318c28a8ae02f0cd2bb999e086a72ef847a310d Mon Sep 17 00:00:00 2001 From: Erik Costlow Date: Wed, 11 Nov 2026 12:11:27 -0500 Subject: [PATCH 1/4] =?UTF-8?q?Steer:=20OpenCode=20receivers=20take=20agen?= =?UTF-8?q?t=20steers=20at=20a=20tool=20boundary=20through=20the=20session?= =?UTF-8?q?=20inbox,=20no=20abort=20(steer-delivery-design.md=20=C2=A73.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal-docs/message-steering-design.md | 10 +- internal-docs/steer-delivery-design.md | 86 ++++++++- server/agent-reference/messaging.md | 2 +- server/backends/opencode/adapter.test.ts | 90 +++++++++ server/backends/opencode/adapter.ts | 13 ++ server/backends/opencode/fake-opencode-v2.ts | 127 +++++++++++++ server/backends/opencode/transport.test.ts | 182 +++++++++++++++++++ server/backends/opencode/transport.ts | 143 +++++++++++++++ 8 files changed, 640 insertions(+), 13 deletions(-) diff --git a/internal-docs/message-steering-design.md b/internal-docs/message-steering-design.md index 0ed1930ba..0fa0ed76f 100644 --- a/internal-docs/message-steering-design.md +++ b/internal-docs/message-steering-design.md @@ -78,11 +78,11 @@ flush), fire-and-forget, exactly as the composer's Ctrl/Cmd+Enter does. | Over the steer rate limit | `{queued:true, steered:false, steerDeclined:"rate_limited"}` | queued | | Deduped retry | `{messageId}` only | nothing touched, nothing interrupted | -Since 2026-09-26 (`steer-delivery-design.md`), a steer at a busy Claude receiver aborts nothing: the next -main-thread tool batch delivers the queue inside the turn, or the turn-end flush does. The ack shape is unchanged; -for those receivers `steered:true` means "delivered at the next tool batch, or as the next turn", and the rate limit -does not count them. Codex and OpenCode receivers keep the abort row above, and every flush after a send-now abort -names the cause of the interruption. +Since 2026-09-26 (`steer-delivery-design.md`), a steer at a busy receiver aborts nothing on any engine: the next +main-thread tool boundary delivers the queue inside the turn, or the turn-end flush does (Claude since +2026-09-26, Codex since 2026-10-19, OpenCode since the 2.2.5 pin on 2026-11-11). The ack shape is unchanged; +for those receivers `steered:true` means "delivered at the next tool boundary, or as the next turn", and the rate +limit does not count them. Every flush after a send-now abort names the cause of the interruption. `queued:false` on a steer is deliberate: `queued` answers "does this wait for their current turn", and a steer exists precisely so it does not. `steered:true` is asserted at interrupt-issue time, not delivery time. diff --git a/internal-docs/steer-delivery-design.md b/internal-docs/steer-delivery-design.md index 430a04754..b3297f00a 100644 --- a/internal-docs/steer-delivery-design.md +++ b/internal-docs/steer-delivery-design.md @@ -4,6 +4,8 @@ > returned the text" (no transcript read), and §5 is settled. Tasks 0a248523 (P0, steer read as a human rejection) and 7529b23b (P3, deliver > at the next tool boundary). Builds on `message-steering-design.md` §2.1, which shipped the `steer` flag. > Codex receivers joined in task 85706f13 (2026-10-19): §3.4 delivers their steers with `turn/steer`. +> OpenCode receivers joined on the `@opencode/cli` 2.2.5 pin (2026-11-11): §3.5 delivers their steers through +> the session's own inbox. Every engine now takes agent steers at a boundary. ## 1. The failure, reproduced @@ -65,6 +67,17 @@ that races the end of the run starts a fresh run on its own (43 of 48 timed race follow-on turn of the July attempt, and the source shows a narrow window where such a prompt gets no reply at all. OpenCode therefore keeps the abort. +Re-probed on the current pin (2026-11-11, `@opencode/cli` 2.2.5 source at tag v2.2.5 plus the pinned binary with +the gate mock; 48 timed race trials per mode): the v2 inbox changes the picture. `POST /api/session/:id/prompt` +durably admits its input to the session's SQLite inbox and answers at once; a busy run promotes rows admitted +with `delivery:"steer"` (the default) at its next step boundary, inside the same run - the tool finishes, no +interrupt. `resume:false` admits WITHOUT scheduling a run (a run starts only through the coordinator's wake), so +a steer that misses the run's last boundary check stays a pending row instead of starting anything: 0 of 48 race +trials produced an unowned run or a lost reply (31 folded into the run, 17 stayed pending and cancelled). The +default path (an ordinary prompt) still starts a successor run when the wake lands in the settle window - 13 of +48 trials - but the no-reply window of 1.18.35 is gone: the row is durable and every drain re-checks it. §3.5 +ships the `resume:false` form. + Probe scripts: `internal-docs/evidence/steer-delivery-0926/` (`probe.ts` for the table above, `probe2.ts` for §3.1). The Codex and OpenCode probes of 2026-10-19 were scratch runs and are not kept. @@ -170,8 +183,9 @@ A real human permission denial followed by an ordinary agent message carries no same turn stays a denial even when a steer follows it. The sender half of that line has since moved to the messaging page of the agent reference -(`server/agent-reference/messaging.md`), which since task 85706f13 reads "a Claude or Codex agent receives it without being -interrupted; an OpenCode agent is interrupted." +(`server/agent-reference/messaging.md`), which since task 85706f13 read "a Claude or Codex agent receives it without being +interrupted; an OpenCode agent is interrupted", and since §3.5 (2026-11-11) names no engine: a busy recipient +receives a steer at a tool boundary without being interrupted, whichever engine runs it. ### 3.4 Codex: agent steers go in with `turn/steer` (85706f13) @@ -210,16 +224,56 @@ first (4 of 4). **Edit and fork.** On Codex the steer is a backend user message inside a turn. Its text starts with the note and the sender prefix, so it never matches a member's text, and `editMessage` already excludes `tool_boundary` entries. +### 3.5 OpenCode: agent steers go in through the session's own inbox (2.2.5 pin, 2026-11-11) + +The OpenCode backend declares the same `toolBoundaryDelivery` capability, so the enqueue branch of §3.1 and the +whole orchestrator half of §3.4 are unchanged: same session gate, eligible-batch rule, `boundaryText`, push at +the steer's enqueue and at each main-thread `tool_call`, rate-limit exemption. OpenCode is a push session like +Codex; its `steer()` rides the v2 inbox that §2's re-probe qualified. + +**Push.** `OpenCodeSession.steer(text, steerId)` (transport.steer) sends +`POST /api/session/:id/prompt` with `{id: "msg_", text, delivery: "steer", resume: false}`. +`resume:false` is the precondition the 1.18.35 probe found missing: the admit is durable, but only the +coordinator's wake starts a run (upstream `core/src/session/session.ts:187`, +`core/src/session/run-coordinator.ts:119-140`), and this call never wakes. A run that is already active promotes +the row at its next step boundary (`core/src/session/runner/llm.ts:95-98,187-199`); a steer that misses the +run's last boundary check stays pending and can never start an unowned run. The caller-chosen inbox id (the +route only requires the `msg_` prefix, `schema/src/session-message.ts:21`) is registered before the POST, so the +delivered frame is recognized even when it outruns the admit's answer. The same guards as Codex answer false +with no request (no active turn, no prompt out yet, abort in flight); a refused admit (any HTTP failure, a 409 +id conflict) answers false and never throws. + +**Delivery is the inbox event.** OpenCode promotes the row inside the run and publishes +`session.inbox.delivered` on the same ordered event stream, always before the run's terminal frame. The +transport maps a delivered frame of one of its own inbox ids to `steer_delivered` (once per id), and the +orchestrator logs the items at that point, as in §3.4. + +**Release.** The transport releases every admitted-but-undelivered steer inside the terminal-frame handling, in +`abort()` after the interrupt (which parks rows: no `resume`), and on the failed turn's stop chain that the next +turn waits out; a pre-prompt cancel catches survivors of a server replacement. The cancel route is no oracle: +2.2.5 answers 204 for every `DELETE /api/session/:id/inbox/:inboxId` of an existing session, swallowing the +conflict of a delivered or unknown row (`server/src/handlers/session.ts`, `session.inbox.cancel`). So the +pending list decides first: `GET /api/session/:id/inbox` still listing the row means the run never took it - +the DELETE cancels it; a row gone from the list was promoted (its delivered frame lost with a broken stream), +which is reported as `steer_delivered` before the completion. The session is idle at every one of those points +(Isomux sends the only wakes), so the list cannot change between the read and the cancel. Cancelled items go +back to the queue with the turn's drain (`drainBoundaryClaim`), and the flush that follows delivers them once. +At-least-once, as §3.1/§3.4: a duplicate remains possible only when the list read or the DELETE itself fails +against a live server, since a row it misses is promoted by the next run before the flush's copy. + +**Edit and fork.** As on Codex, the steer is a backend user message inside the turn; its text starts with the +note and the sender prefix, and `editMessage` already excludes `tool_boundary` entries (§3.1). + ## 4. What changes for callers | Surface | Before | After | |---|---|---| -| Ack `steered:true` | turn aborted | Claude receiver: delivered at its next batch boundary, or as its next turn if the turn ends first or the queue holds a member/skill/handoff item. Codex receiver (since 85706f13): delivered at its next step (after the running tool or reply, or at the next poll of a long command, which keeps running), or as its next turn if codex refuses it or drops it. OpenCode: turn aborted (unchanged). Shape unchanged. | -| Ack `queued:false` on a steer | "does not wait for the current turn" | Claude or Codex receiver: can wait for the rest of the turn (long tool call, no further boundary, ineligible batch) | +| Ack `steered:true` | turn aborted | Claude receiver: delivered at its next batch boundary, or as its next turn if the turn ends first or the queue holds a member/skill/handoff item. Codex receiver (since 85706f13): delivered at its next step (after the running tool or reply, or at the next poll of a long command, which keeps running), or as its next turn if codex refuses it or drops it. OpenCode receiver (since the 2.2.5 pin, §3.5): as Codex - delivered at the run's next step boundary, or as its next turn if the run ended first or OpenCode refused the admit. Shape unchanged. | +| Ack `queued:false` on a steer | "does not wait for the current turn" | Claude, Codex or OpenCode receiver: can wait for the rest of the turn (long tool call, no further boundary, ineligible batch) | | Rate limit (3/min/receiver) | counts steers that interrupted | unchanged meaning: boundary steers interrupt nothing, so they are neither counted nor refused | -| Long tool call at the receiver (for example a 15-minute suite) | steer cuts it | Claude: steer waits for it. Codex: the command keeps running and the steer can go in at its next poll. OpenCode: steer cuts it | +| Long tool call at the receiver (for example a 15-minute suite) | steer cuts it | Claude: steer waits for it. Codex: the command keeps running and the steer can go in at its next poll. OpenCode: the tool runs to its end and the steer goes in at the step boundary after it | | Ordinary agent that must stop a Claude agent now | steer | no path; privileged agents and members keep `/abort` and Send now. Since task f4452169 (Nil, 2026-10-02) every agent may `/abort` any agent it may message; agent stops share the steer window and leave the target a stop notice | -| Chat log | "Agent interrupted." then the message | no interruption line. Claude: the message appears after the tool result. Codex: where codex recorded it, which for a long command can be between the command card and its result | +| Chat log | "Agent interrupted." then the message | no interruption line. Claude: the message appears after the tool result. Codex: where codex recorded it, which for a long command can be between the command card and its result. OpenCode: after the tool result of the step whose boundary promoted it | | Curl card label for a steer | "Interrupt {who} with a message" | "Message {who} mid-turn" (en; es, ca, zh follow) | ## 5. Decisions (PM, 2026-09-26) @@ -228,7 +282,9 @@ the sender prefix, so it never matches a member's text, and `editMessage` alread privileged agents keep `/abort`. The PM takes the contract change to Nil. 2. The rate limit counts only steers that abort. 3. Codex `turn/steer`: a follow-up task. Codex and OpenCode keep the abort plus the §3.2 note. Done for Codex in - task 85706f13 (§3.4, 2026-10-19); OpenCode keeps the abort (§2). + task 85706f13 (§3.4, 2026-10-19); done for OpenCode on the 2.2.5 pin (§3.5, 2026-11-11) after the §2 + re-probe qualified the native inbox steer. An agent steer now aborts no engine; the §3.2 notes remain for + the abort paths that are left (a member's Send now, an agent's `/abort`, and the send-now route). ## 6. Verification @@ -269,3 +325,19 @@ the sender prefix, so it never matches a member's text, and `editMessage` alread recorded the text at a poll boundary; the chat showed it before the Bash card, which the adapter reported only when the command ended until task 9f8d6942); the Receiver did steps 2 and 3 and replied with the line count. It asked nobody anything and did not treat the steer as injected text. +- Unit, OpenCode push (§3.5; `transport.test.ts` "OpenCode 2 native steer", `adapter.test.ts` "OpenCode session + steer"): an admitted steer folds into the running turn - `steer_delivered` before `turn_completed`, a foreign + inbox row's delivery reports nothing, no cancel owed; false with no request when no turn runs; a refused admit + answers false and leaves nothing pending; a parked steer is cancelled before the completion is emitted; a row + gone from the pending list (the cancel route answers 204 for both outcomes) is reported as the delivery, once, + before the completion, with no DELETE; abort cancels a parked steer after the interrupt; the backend declares + the capability; the session's stream surfaces the delivery. The fake answers the steer route (`resume:false` + prompts), the inbox GET/DELETE routes (204 for every cancel, as 2.2.5), and plays each turn's boundary as + `deliver`, `silent` (a frame the stream lost) or `park` (the steer missed the run). +- Probe, OpenCode 2.2.5 (§2 re-probe, 2026-11-11): pinned darwin-arm64 binary with the gate mock, 48 timed race + trials per mode. `resume:false`: 31 folded into the run, 17 stayed pending and cancelled (204), 0 successor + runs, 0 lost. Default prompts: 35 folded, 13 successor runs, 0 lost. Mid-run steer during a 6 s tool: the tool + ran to its end, one run, the next model request carried the text. Idle admit with `resume:false`: no execution + started while the row was pending. Scratch runs, not kept. +- Unverified for OpenCode: a pending row across an OpenCode server restart (durable in its SQLite by design, and + the next turn's pre-prompt cancel guards it); the §1 live repro at an OpenCode receiver. diff --git a/server/agent-reference/messaging.md b/server/agent-reference/messaging.md index 7ee46fda0..2642284b5 100644 --- a/server/agent-reference/messaging.md +++ b/server/agent-reference/messaging.md @@ -1,6 +1,6 @@ # Messaging -Send to another agent with `POST /api/agents/:id/messages`: the body is the message as `text/plain`, and options go in the query: `steer=1`, `clientMessageId=`. The token supplies sender identity. A response with `queued:true` waits for the current turn to finish; an idle recipient receives it now. To reach a busy recipient during its current turn, add `steer=1`: a Claude or Codex agent receives it without being interrupted; an OpenCode agent is interrupted. When starting an exchange, make sure one side steers and tell the peer. Otherwise both sides can continue on stale information. `clientMessageId` makes retries safe for five minutes. +Send to another agent with `POST /api/agents/:id/messages`: the body is the message as `text/plain`, and options go in the query: `steer=1`, `clientMessageId=`. The token supplies sender identity. A response with `queued:true` waits for the current turn to finish; an idle recipient receives it now. To reach a busy recipient during its current turn, add `steer=1`: the recipient receives it without being interrupted, as soon as its running tool calls end. When starting an exchange, make sure one side steers and tell the peer. Otherwise both sides can continue on stale information. `clientMessageId` makes retries safe for five minutes. To stop another agent's current turn without a message, use `POST /api/agents/:id/abort` with `{}`. It also denies a permission prompt the agent is parked on. Every agent can stop an agent it can message. diff --git a/server/backends/opencode/adapter.test.ts b/server/backends/opencode/adapter.test.ts index eff5c15b5..5fd368ce6 100644 --- a/server/backends/opencode/adapter.test.ts +++ b/server/backends/opencode/adapter.test.ts @@ -19,6 +19,7 @@ import { import { STATE_ROOT } from "../../config.ts"; import { AGENT_ROOT } from "../../split/roots.ts"; import { + FAKE_SESSION, fakeOpenCode, fakeSupervisor, fakeToolEnvironment, @@ -1546,3 +1547,92 @@ describe("OpenCode pinned transport", () => { ); } }); + +// Agent steers reach a busy OpenCode session through its inbox, without an +// abort (internal-docs/steer-delivery-design.md §3.5). The orchestrator's +// push path (message-queue pushBoundarySteer) reads the capability flag and +// the session's steer(); the transport covers the wire behavior +// (transport.test.ts, "OpenCode 2 native steer"). +describe("OpenCode session steer", () => { + it("the OpenCode backend delivers agent steers without aborting", () => { + // The enqueue path reads this flag; without it every steer still aborts. + expect(getBackend("opencode").toolBoundaryDelivery).toBe(true); + }); + + it("steers the running turn and surfaces the delivery on its stream", async () => { + const f = framesOf(FAKE_SESSION); + const turn: FakeTurn = { + frames: [ + f.delivered("INBOX"), + f.stepStarted("m1"), + ...f.toolCalled("m1", "call_1", "shell", { command: "sleep 5" }), + f.asked("per_1", "sleep 5"), + f.toolSuccess("m1", "call_1", "slept"), + f.stepEnded("m1"), + f.textEnded("m1", "done"), + f.succeeded(), + ] as Frame[], + session: FAKE_SESSION, + inbox: "INBOX", + }; + const fake = fakeOpenCode({ turns: [turn] }); + cleanup.push(fake.restore); + const backend = createOpenCodeBackend({ + supervisor: fakeSupervisor().supervisor, + }); + const session = backend.createSession({ + ...opts, + modelFamily: "provider/model", + }); + cleanup.push(() => session.close()); + + const events: NormalizedEvent[] = []; + const streamed = (async () => { + for await (const event of session.stream()) { + events.push(event); + if (event.kind === "turn_completed") return event; + } + return undefined; + })(); + const sending = session.send("work"); + const askAt = Date.now() + 3_000; + while ( + !events.some((e) => e.kind === "approval_request") && + Date.now() < askAt + ) + await Bun.sleep(2); + const ask = events.find( + (e) => e.kind === "approval_request", + ) as Extract; + expect(ask).toBeDefined(); + expect(await session.steer!("FYI from Sender", "steer-1")).toBe(true); + await session.approve(ask.approvalId, { kind: "allow_once" }); + await sending; + const end = await streamed; + + expect(end?.status).toBe("completed"); + const delivered = events.filter((e) => e.kind === "steer_delivered"); + expect(delivered).toEqual([ + { kind: "steer_delivered", steerId: "steer-1" }, + ]); + expect(events.indexOf(delivered[0]!)).toBeLessThan( + events.findIndex((e) => e.kind === "turn_completed"), + ); + expect(fake.inboxRow("msg_steer-1")?.state).toBe("delivered"); + }); + + it("refuses a steer with no turn running", async () => { + const fake = fakeOpenCode({}); + cleanup.push(fake.restore); + const backend = createOpenCodeBackend({ + supervisor: fakeSupervisor().supervisor, + }); + const session = backend.createSession({ + ...opts, + modelFamily: "provider/model", + }); + cleanup.push(() => session.close()); + expect(await session.steer!("early", "steer-0")).toBe(false); + expect(fake.steerLog).toEqual([]); + }); +}); diff --git a/server/backends/opencode/adapter.ts b/server/backends/opencode/adapter.ts index 3169db839..6a0e6328c 100644 --- a/server/backends/opencode/adapter.ts +++ b/server/backends/opencode/adapter.ts @@ -323,6 +323,15 @@ class OpenCodeServerSession implements BackendSession { await this.transport.abort(); } + // Push delivery of an agent steer (message-queue pushBoundarySteer): the + // transport admits it into OpenCode's inbox with resume:false, so it folds + // into the running turn or waits to be cancelled - it never starts a run of + // its own. Delivery is the inbox event, reported as steer_delivered. + async steer(text: string, steerId: string): Promise { + if (this.ended) return false; + return this.transport.steer(text, steerId); + } + canAbortInPlace(): boolean { return this.transport.canAbortInPlace(); } @@ -616,6 +625,10 @@ export function createOpenCodeBackend( }; return { capabilities: CAPABILITIES, + // Agent steers ride the session's own inbox (OpenCodeSession.steer) into + // the running turn instead of aborting it, as on Claude and Codex + // (internal-docs/steer-delivery-design.md §3.5). + toolBoundaryDelivery: true, getModelOptions: () => [], getPermissionModes: () => PERMISSION_MODES, // A saved setting reaches both of the member's servers: the limited one diff --git a/server/backends/opencode/fake-opencode-v2.ts b/server/backends/opencode/fake-opencode-v2.ts index e1ea78201..66d9bffe5 100644 --- a/server/backends/opencode/fake-opencode-v2.ts +++ b/server/backends/opencode/fake-opencode-v2.ts @@ -123,6 +123,13 @@ export interface FakeTurn { interruptBefore?: (frame: Frame) => boolean; // false: play on without waiting for each permission reply. waitForReplies?: boolean; + // What the turn's boundary does with steers still pending when its + // terminal frame goes out (2.2.5 promotes them before the frame): + // "deliver" emits their session.inbox.delivered (the default, the mid-run + // fold-in), "silent" marks them delivered without a frame (a frame the + // stream lost; the cancel then answers 409), "park" leaves them pending + // (the steer missed the run's last boundary check). + steers?: "deliver" | "silent" | "park"; } export function fixtureTurn( @@ -148,6 +155,9 @@ type Route = | "session.list" | "messages" | "prompt" + | "steer" + | "inbox.list" + | "inbox.cancel" | "instructions" | "model.select" | "interrupt" @@ -317,6 +327,37 @@ export function fakeOpenCode(options: FakeOptions = {}) { const agents = new Map(); // Every prompt, with the agent its session had when it arrived. const promptLog: { session: string; text: unknown; agent: string }[] = []; + // The inbox rows steer prompts (resume:false) admitted, by inbox id. A row + // is pending until the turn's boundary delivers it (play's flush) or a + // DELETE cancels it, as in 2.2.5's session_inbox table. + const inboxRows = new Map< + string, + { + id: string; + sessionID: string; + text: unknown; + state: "pending" | "delivered" | "cancelled"; + } + >(); + // Every steer prompt, admitted or not. + const steerLog: { session: string; id: unknown; text: unknown }[] = []; + // The boundary flush of one turn: what happens to its session's pending + // steers when its terminal frame goes out. + const flushSteers = ( + session: string, + mode: "deliver" | "silent" | "park", + ) => { + if (mode === "park") return; + for (const row of inboxRows.values()) { + if (row.sessionID !== session || row.state !== "pending") continue; + row.state = "delivered"; + if (mode === "deliver") + emit({ + type: "session.inbox.delivered", + data: { sessionID: session, inboxID: row.id }, + }); + } + }; const playing: Promise[] = []; const encoder = new TextEncoder(); // Frames emitted while no stream is open are lost, as on /api/event. @@ -387,6 +428,9 @@ export function fakeOpenCode(options: FakeOptions = {}) { }); for (const frame of frames.slice(emitFrom)) { if (turn.interruptBefore?.(frame)) await interrupted(); + // 2.2.5 promotes pending steers at the boundary before the run ends. + if (TERMINAL.has(frame.type)) + flushSteers(session, turn.steers ?? "deliver"); emit(frame); await tick(); if (frame.type === "permission.asked" && turn.waitForReplies !== false) @@ -617,6 +661,56 @@ export function fakeOpenCode(options: FakeOptions = {}) { cursor: { ...(next ? { next } : {}) }, }); }); + // A steer admit (2.2.5's prompt with resume:false): a durable inbox row + // the running turn promotes at its next boundary, no run scheduled, no + // turn consumed. The caller's own id, echoed back, 409 on a duplicate. + if ( + method === "POST" && + rest[2] === "prompt" && + (bodyOf(init) as { resume?: unknown } | undefined)?.resume === false + ) + return r("steer", () => { + const body = bodyOf(init) as + | { id?: unknown; text?: unknown; delivery?: unknown } + | undefined; + const inboxId = + typeof body?.id === "string" && body.id + ? body.id + : `msg_steer_${inboxRows.size + 1}`; + steerLog.push({ session: id, id: body?.id, text: body?.text }); + if (inboxRows.has(inboxId)) + return json( + { + _tag: "Session.PromptConflictError", + message: `Prompt message ID conflicts: ${inboxId}`, + }, + 409, + ); + inboxRows.set(inboxId, { + id: inboxId, + sessionID: id, + text: body?.text, + state: "pending", + }); + emit({ + type: "session.inbox.enqueued", + data: { + sessionID: id, + inboxID: inboxId, + item: { type: "user", delivery: body?.delivery ?? "steer" }, + }, + }); + return json({ + data: { + id: inboxId, + sessionID: id, + time: { created: 1 }, + type: "user", + payload: { text: body?.text }, + delivery: body?.delivery ?? "steer", + }, + }); + }); if (method === "POST" && rest[2] === "prompt") return r("prompt", async () => { promptCount++; @@ -649,6 +743,35 @@ export function fakeOpenCode(options: FakeOptions = {}) { }, }); }); + // The pending inbox rows only: a delivered or cancelled one is gone from + // the table, as 2.2.5's list answers. + if (method === "GET" && rest[2] === "inbox" && rest.length === 3) + return r("inbox.list", () => + json({ + data: [...inboxRows.values()] + .filter((row) => row.sessionID === id && row.state === "pending") + .map((row) => ({ + id: row.id, + sessionID: row.sessionID, + time: { created: 1 }, + type: "user", + payload: { text: row.text }, + delivery: "steer", + })), + }), + ); + // 2.2.5 answers 204 for every cancel of an existing session: a + // delivered, cancelled or unknown row is a lifecycle conflict the + // handler swallows (server/src/handlers/session.ts, + // session.inbox.cancel), so the pending list is the only way to tell + // what a cancel did. + if (method === "DELETE" && rest[2] === "inbox" && rest.length === 4) + return r("inbox.cancel", () => { + const row = inboxRows.get(rest[3]); + if (row && row.sessionID === id && row.state === "pending") + row.state = "cancelled"; + return new Response(null, { status: 204 }); + }); // 2.1.4 changes a session's agent here; PATCH ignores an agent field // (2026-10-25). if (method === "POST" && rest[2] === "agent") @@ -754,6 +877,10 @@ export function fakeOpenCode(options: FakeOptions = {}) { ), replies, promptLog, + // Every steer prompt that arrived, admitted or not. + steerLog, + // The inbox row of a steer id: its admission state, or undefined. + inboxRow: (inboxId: string) => inboxRows.get(inboxId), // The shells DELETE /api/shell/:id removed, in order. deletedShells: () => [...deletedShells], interrupts: () => interrupts, diff --git a/server/backends/opencode/transport.test.ts b/server/backends/opencode/transport.test.ts index de743ee17..a1784e378 100644 --- a/server/backends/opencode/transport.test.ts +++ b/server/backends/opencode/transport.test.ts @@ -1809,6 +1809,188 @@ describe("OpenCode 2 permissions", () => { }); }); +// Agent steers ride OpenCode's own inbox (internal-docs/steer-delivery-design.md +// §3.5, probed on 2.2.5): POST prompt with delivery:"steer" and resume:false +// admits a durable row the running turn promotes at its next step boundary - +// no interrupt, no run of its own. A steer the run never picked up is +// cancelled before the turn ends, so the orchestrator's flush can send it +// itself; a row gone from the pending list means OpenCode promoted it after +// all (the cancel route answers 204 for both). +describe("OpenCode 2 native steer", () => { + it("folds an admitted steer into the running turn and reports it before the turn ends", async () => { + const h = setup({ fake: { turns: [fixtureTurn("shell-allow")] } }); + const sending = h.turn([{ type: "text", text: "go" }], (event) => { + if (event.kind === "approval_request") + // Not ours: a delivery of some other inbox row reports nothing. + h.fake.emit({ + type: "session.inbox.delivered", + data: { sessionID: h.fake.sessionId, inboxID: "msg_foreign" }, + }); + }); + // The steer admits while the ask parks the turn; the approve lets it run + // to its boundary, where 2.2.5 promotes the row. + await until(() => h.events.some((e) => e.kind === "approval_request")); + expect(await h.transport.steer("FYI from Sender", "steer-a")).toBe(true); + const ask = h.events.find( + (e) => e.kind === "approval_request", + ) as Extract; + await h.transport.approve(ask.approvalId, { kind: "allow_once" }); + const { events, end } = await sending; + + expect(end?.status).toBe("completed"); + // The admit is one prompt call with the caller's own id, steer delivery + // and no run scheduled - the precondition that keeps a late steer from + // starting an unowned run. + expect(h.fake.steerLog).toEqual([ + { session: h.fake.sessionId, id: "msg_steer-a", text: "FYI from Sender" }, + ]); + const steerBody = h.fake + .calls("POST", /\/prompt$/) + .map((r) => r.body) + .filter((body) => (body as { resume?: unknown })?.resume === false); + expect(steerBody).toEqual([ + { + id: "msg_steer-a", + text: "FYI from Sender", + delivery: "steer", + resume: false, + }, + ]); + // The boundary delivered it: one steer_delivered, before the completion, + // and nothing for the foreign row. No cancel was owed. + const delivered = events.filter((e) => e.kind === "steer_delivered"); + expect(delivered).toEqual([{ kind: "steer_delivered", steerId: "steer-a" }]); + expect(events.indexOf(delivered[0]!)).toBeLessThan( + events.findIndex((e) => e.kind === "turn_completed"), + ); + expect(h.fake.inboxRow("msg_steer-a")?.state).toBe("delivered"); + expect(h.fake.calls("DELETE", /\/inbox\//)).toEqual([]); + }); + + it("answers false with no request when no turn is running", async () => { + const h = setup({ fake: { turns: [fixtureTurn("shell-allow")] } }); + // Before any turn. + expect(await h.transport.steer("early", "steer-e")).toBe(false); + const { end } = await h.turn([{ type: "text", text: "go" }], (event) => { + if (event.kind === "approval_request") + void h.transport.approve(event.approvalId, { kind: "allow_once" }); + }); + expect(end?.status).toBe("completed"); + // After the turn ended. + expect(await h.transport.steer("late", "steer-l")).toBe(false); + expect(h.fake.steerLog).toEqual([]); + }); + + it("answers false, never throws, when OpenCode refuses the admit", async () => { + const h = setup({ + fake: { status: { steer: 500 }, turns: [fixtureTurn("shell-allow")] }, + }); + const sending = h.turn([{ type: "text", text: "go" }]); + await until(() => h.events.some((e) => e.kind === "approval_request")); + expect(await h.transport.steer("refused", "steer-r")).toBe(false); + const ask = h.events.find( + (e) => e.kind === "approval_request", + ) as Extract; + await h.transport.approve(ask.approvalId, { kind: "allow_once" }); + const { events, end } = await sending; + expect(end?.status).toBe("completed"); + // The admit went out and was refused (the fake's status short-circuits + // its route answer, so the request count is the evidence). + expect(h.fake.count("steer")).toBe(1); + // The refusal left nothing pending: no delivery, no cancel. + expect(events.filter((e) => e.kind === "steer_delivered")).toEqual([]); + expect(h.fake.calls("DELETE", /\/inbox\//)).toEqual([]); + }); + + it("cancels a steer the turn's end left pending, before the completion", async () => { + const h = setup({ + fake: { turns: [fixtureTurn("shell-allow", 0, { steers: "park" })] }, + }); + let cancelsAtEnd = -1; + const sending = h.turn([{ type: "text", text: "go" }], (event) => { + if (event.kind === "turn_completed") + cancelsAtEnd = h.fake.calls("DELETE", /\/inbox\//).length; + }); + await until(() => h.events.some((e) => e.kind === "approval_request")); + expect(await h.transport.steer("too late", "steer-p")).toBe(true); + const ask = h.events.find( + (e) => e.kind === "approval_request", + ) as Extract; + await h.transport.approve(ask.approvalId, { kind: "allow_once" }); + const { events, end } = await sending; + + expect(end?.status).toBe("completed"); + // The cancel ran before the completion went out, so the flush the + // completion releases cannot double-deliver the row's text. + expect(cancelsAtEnd).toBe(1); + expect(h.fake.calls("DELETE", /\/inbox\/msg_steer-p$/)).toHaveLength(1); + expect(h.fake.inboxRow("msg_steer-p")?.state).toBe("cancelled"); + expect(events.filter((e) => e.kind === "steer_delivered")).toEqual([]); + }); + + it("reads a row gone from the inbox as the delivery the stream lost", async () => { + const h = setup({ + fake: { turns: [fixtureTurn("shell-allow", 0, { steers: "silent" })] }, + }); + const sending = h.turn([{ type: "text", text: "go" }]); + await until(() => h.events.some((e) => e.kind === "approval_request")); + expect(await h.transport.steer("quietly taken", "steer-s")).toBe(true); + const ask = h.events.find( + (e) => e.kind === "approval_request", + ) as Extract; + await h.transport.approve(ask.approvalId, { kind: "allow_once" }); + const { events, end } = await sending; + + expect(end?.status).toBe("completed"); + // The cancel answered 204 for everything, so the pending list decided: + // the row OpenCode promoted is gone from it, and no DELETE went out. + expect(h.fake.inboxRow("msg_steer-s")?.state).toBe("delivered"); + expect(h.fake.calls("GET", /\/inbox$/)).toHaveLength(1); + expect(h.fake.calls("DELETE", /\/inbox\//)).toEqual([]); + // The promotion means the model read the steer: the delivery is reported + // before the completion, exactly once, so the flush does not resend it. + const delivered = events.filter((e) => e.kind === "steer_delivered"); + expect(delivered).toEqual([{ kind: "steer_delivered", steerId: "steer-s" }]); + expect(events.indexOf(delivered[0]!)).toBeLessThan( + events.findIndex((e) => e.kind === "turn_completed"), + ); + }); + + it("cancels a parked steer on abort, after the interrupt", async () => { + const h = setup({ + fake: { + turns: [ + fixtureTurn("shell-deny", 0, { + waitForReplies: false, + interruptBefore: (frame) => frame.type === "permission.replied", + }), + ], + }, + }); + let steered: boolean | undefined; + const sending = h.turn([{ type: "text", text: "go" }], (event) => { + if (event.kind === "approval_request") + void (async () => { + steered = await h.transport.steer("parked", "steer-x"); + await h.transport.abort(); + })(); + }); + const { events, end } = await sending; + expect(steered).toBe(true); + expect(end?.status).toBe("interrupted"); + const interrupt = h.fake.requests.findIndex((r) => + r.path.endsWith("/interrupt"), + ); + const cancel = h.fake.requests.findIndex( + (r) => r.method === "DELETE" && /\/inbox\/msg_steer-x$/.test(r.path), + ); + expect(interrupt).toBeGreaterThan(-1); + expect(cancel).toBeGreaterThan(interrupt); + expect(h.fake.inboxRow("msg_steer-x")?.state).toBe("cancelled"); + expect(events.filter((e) => e.kind === "steer_delivered")).toEqual([]); + }); +}); + describe("OpenCode 2 failures (T10)", () => { // Task 2008c993: a refusal of OpenCode's own Zen or Go plan is a usage // limit stop, which the office fallback acts on. It names no reset time, and diff --git a/server/backends/opencode/transport.ts b/server/backends/opencode/transport.ts index 84e3ab237..7d4614fc5 100644 --- a/server/backends/opencode/transport.ts +++ b/server/backends/opencode/transport.ts @@ -587,6 +587,14 @@ export class OpenCodeTransport { // Open permission requests: request id -> OpenCode session id. OpenCode // runs the tool calls of one step in parallel, and each can ask at once. private pendingPermissions = new Map(); + // Agent steers admitted into OpenCode's inbox that it has not delivered + // yet: the inbox id (msg_, chosen here so the delivered frame is + // recognized even when it outruns the admit's answer) -> the orchestrator's + // steer id and the admit, which a cancel waits on so it never races it. + private readonly pendingSteers = new Map< + string, + { steerId: string; admitted: Promise } + >(); // The open turn's question forms: form id -> session id. private openForms = new Map(); // The open turn's subagent sessions, nested ones included. @@ -922,6 +930,9 @@ export class OpenCodeTransport { () => undefined, () => undefined, ) + // A steer this turn admitted must not outlive it; the next turn + // waits this chain out before its prompt. + .then(() => this.cancelPendingSteers()) .then(() => this.sweepLeakedShells(scope)) .finally(() => { if (this.sessionStop === stop) this.sessionStop = null; @@ -1110,6 +1121,11 @@ export class OpenCodeTransport { if (settled) return; await this.previousSessionStop(); if (settled) return; + // A steer row a previous turn's cancel could not reach (its server was + // replaced mid-turn) must not ride into this run: OpenCode promotes + // pending steers before the input they were never claimed with. + await this.cancelPendingSteers(); + if (settled) return; // A stop during the waits above: no prompt is out, so none goes. if (this.abortRequested) { emit({ kind: "turn_completed", status: "interrupted" }); @@ -1227,6 +1243,117 @@ export class OpenCodeTransport { } } + // An agent steer into the running turn, through OpenCode's own inbox + // (internal-docs/steer-delivery-design.md §3.5, probed on 2.2.5). + // delivery:"steer" makes the run promote the text at its next step + // boundary: the running tool finishes and the next model request carries + // the steer, with no interrupt. resume:false admits WITHOUT scheduling a + // run, so a steer that misses the run's last boundary check stays a + // pending inbox row instead of starting an unowned run; the turn's end + // cancels it and the orchestrator's flush delivers it. Answers false, + // never throws, when no turn is running or OpenCode refused the admit. + async steer(text: string, steerId: string): Promise { + if ( + !this.activeTurn || + !this.promptSent || + this.abortRequested || + !this.sessionId + ) + return false; + const inboxId = `msg_${steerId}`; + const admitted = this.admitSteer(inboxId, text); + this.pendingSteers.set(inboxId, { steerId, admitted }); + if (await admitted) return true; + const entry = this.pendingSteers.get(inboxId); + if (entry?.admitted === admitted) this.pendingSteers.delete(inboxId); + return false; + } + + private async admitSteer(inboxId: string, text: string): Promise { + if (!this.sessionId) return false; + try { + const response = await this.request( + `/api/session/${encodeURIComponent(this.sessionId)}/prompt`, + { + method: "POST", + body: JSON.stringify({ + id: inboxId, + text, + delivery: "steer", + resume: false, + }), + }, + ); + const data = asRecord(asRecord(await response.json()).data); + // The admit must carry the id this transport chose: the delivered + // frame and the cancel both key on it. A duplicate id is a 409 the + // catch below turns into a refusal. + if (stringField(data, "id") !== inboxId) return false; + this.contractShapeSink?.("http:prompt:{data:{id:string}}"); + return true; + } catch { + return false; + } + } + + // Every steer this turn admitted that OpenCode has not delivered: cancel + // its inbox row, so no later run promotes it after the orchestrator + // released the items to the flush (a duplicate for the model). The cancel + // route answers 204 for every row of an existing session - it swallows the + // conflict of a delivered or unknown one (2.2.5 + // server/src/handlers/session.ts, session.inbox.cancel) - so the pending + // list decides first: a row still listed is cancelled; a row gone from it + // was promoted into the run, which is the delivery this stream missed and + // is reported as such. Runs before the turn's completion is emitted + // (terminal frames, abort) and on the failed turn's stop chain, which the + // next turn waits out before its own prompt. The session is idle at every + // one of those points (Isomux sends the only wakes), so the list cannot + // change between the read and the cancel. + private async cancelPendingSteers(): Promise { + if (this.pendingSteers.size === 0) return; + const entries = [...this.pendingSteers.entries()]; + this.pendingSteers.clear(); + const sessionId = this.sessionId; + if (!sessionId) return; + const session = encodeURIComponent(sessionId); + await Promise.all( + entries.map(async ([inboxId, entry]) => { + if (!(await entry.admitted)) return; + let pending = false; + try { + const list = listData( + await this + .request(`/api/session/${session}/inbox`) + .then((r) => r.json()), + ); + pending = list.some( + (row) => stringField(asRecord(row), "id") === inboxId, + ); + } catch { + // A list this turn cannot read leaves the items to the flush. + return; + } + if (!pending) { + this.turnSink?.({ + kind: "steer_delivered", + steerId: entry.steerId, + }); + return; + } + try { + await this.request( + `/api/session/${session}/inbox/${encodeURIComponent(inboxId)}`, + { method: "DELETE" }, + ); + } catch { + // Best effort: a row this cancel missed is promoted by the next + // run, which the flush's copy then duplicates (at-least-once, the + // policy of §3.1). + } + }), + ); + } + async abort(): Promise { // No prompt is out during a pre-stream wait, so OpenCode has nothing to // interrupt, and a request to this folder would wait on its setup too. @@ -1242,6 +1369,9 @@ export class OpenCodeTransport { `/api/session/${encodeURIComponent(this.sessionId)}/interrupt`, { method: "POST" }, ).catch(() => undefined); + // A parked steer must not outlive the turn into the next run: cancel it, + // so the orchestrator's flush delivers its items itself. + await this.cancelPendingSteers(); } async approve(approvalId: string, decision: ApprovalDecision): Promise { @@ -1884,6 +2014,10 @@ export class OpenCodeTransport { }; // An end of this turn's busy period, mapped to the turn's end. const terminal = async (frame: Record): Promise => { + // A steer the run never picked up is cancelled before the completion + // reaches the orchestrator, whose flush would send its items again. + // A cancel that finds the row promoted reports the delivery first. + await this.cancelPendingSteers(); const data = asRecord(frame.data); if (frame.type === "session.execution.succeeded") { settle({ kind: "turn_completed", status: "completed" }); @@ -2055,6 +2189,15 @@ export class OpenCodeTransport { if (type === "session.inbox.delivered") { const inbox = stringField(data, "inboxID"); if (!inbox) continue; + // A steer this transport admitted: OpenCode promoted it into + // the run, so the model reads it before its next request. The + // orchestrator logs its items on this event. + const steered = this.pendingSteers.get(inbox); + if (steered !== undefined) { + this.pendingSteers.delete(inbox); + sink({ kind: "steer_delivered", steerId: steered.steerId }); + continue; + } if (turn.id === null) turn.early.push({ kind: "delivered", id: inbox, at: created }); else if (inbox === turn.id && !turn.armed) { From 97e947ce2bf264e397272aff2f817e52250ae95e Mon Sep 17 00:00:00 2001 From: Erik Costlow Date: Wed, 11 Nov 2026 13:08:46 -0500 Subject: [PATCH 2/4] =?UTF-8?q?Tests:=20live=20steer=20cases=20against=20t?= =?UTF-8?q?he=20real=20pinned=20OpenCode=202.2.5=20-=20fold-in=20at=20the?= =?UTF-8?q?=20boundary,=20the=20cancelled=20miss,=20abort=20with=20a=20par?= =?UTF-8?q?ked=20steer=20(steer-delivery-design.md=20=C2=A73.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../agent-manager.opencode.live.test.ts | 563 ++++++++++++++++++ 1 file changed, 563 insertions(+) diff --git a/server/test-support/agent-manager.opencode.live.test.ts b/server/test-support/agent-manager.opencode.live.test.ts index 2b558b906..ac5e97cbb 100644 --- a/server/test-support/agent-manager.opencode.live.test.ts +++ b/server/test-support/agent-manager.opencode.live.test.ts @@ -384,3 +384,566 @@ it.skipIf(!LIVE)( }, 25_000, ); + +// --------------------------------------------------------------------------- +// Native steer against the real pinned 2.2.5 server (steer-delivery-design.md +// §3.5, probed 2026-11-11): a steer folded into a running turn at the step +// boundary, a steer that misses the run's last boundary check (cancelled, +// then delivered once by the flush), and an abort with a parked steer. Mock +// provider, no model tokens. The cancel race of the miss case is a real +// millisecond window: the test sweeps steer delays until one attempt lands +// in it, asserting the invariants (delivered exactly once, inbox drained, +// no unowned run) on every attempt. +// --------------------------------------------------------------------------- + +const STEER_SENDER = { + kind: "agent" as const, + agentId: "agent-live-sender", + agentName: "Sender", + roomName: "room-live-steer", +}; + +interface SteerMockState { + // Resolved when the mock starts streaming a FAST-ANSWER reply: the run's + // last boundary checks are milliseconds away. + onFinal?: () => void; +} + +// The tool ran to its end: its stdout, echoed in the tool result. (The +// session's cwd is the state root, where the safety plugin denies writes, so +// completion is proven through stdout, not a file.) +const SLEPT_TO_END = "SLEPT-TO-END"; + +// The deterministic provider mock: keyed on the LAST user message, so one +// session's accumulated history never changes a later turn's answer. A +// LONG-SHELL prompt gets one shell tool call (sleep 4, then the stdout +// marker); anything holding steer text gets a fixed reply; the rest a plain +// answer. The shell tool asks (Isomux's BASE_RULES), as in production. +function steerMock(state: SteerMockState) { + // topic: the session's asynchronous topic-label generation, which can + // outlive the turn that triggered it; the quiescence checks ignore it. + const requests: { + messages: Array>; + topic: boolean; + }[] = []; + let callId = 0; + const mock = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + const url = new URL(request.url); + if (url.pathname === "/v1/models") + return Response.json({ + object: "list", + data: [{ id: "gate-model", object: "model" }], + }); + if (url.pathname !== "/v1/chat/completions") + return new Response("not found", { status: 404 }); + const body = (await request.json()) as { + messages?: Array>; + }; + const messages = body.messages ?? []; + const lastUser = [...messages] + .reverse() + .find((message) => message.role === "user")?.content; + const prompt = + typeof lastUser === "string" ? lastUser : JSON.stringify(lastUser ?? ""); + requests.push({ messages, topic: prompt.includes("topic label") }); + const hasToolResult = messages.some((message) => message.role === "tool"); + const id = `gate-${requests.length}`; + const stream = new ReadableStream({ + start(controller) { + const send = (value: unknown) => + controller.enqueue( + `data: ${typeof value === "string" ? value : JSON.stringify(value)}\n\n`, + ); + const chunk = (delta: unknown, finish: string | null) => ({ + id, + object: "chat.completion.chunk", + created: 1, + model: "gate-model", + choices: [{ index: 0, delta, finish_reason: finish }], + }); + send(chunk({ role: "assistant" }, null)); + if (prompt.includes("LONG-SHELL") && !hasToolResult) { + send( + chunk( + { + tool_calls: [ + { + index: 0, + id: `call_${++callId}`, + type: "function", + function: { + name: "shell", + arguments: JSON.stringify({ + command: `sleep 4; echo ${SLEPT_TO_END}`, + }), + }, + }, + ], + }, + null, + ), + ); + send(chunk({}, "tool_calls")); + } else { + if (prompt.includes("FAST-ANSWER")) state.onFinal?.(); + const text = prompt.includes("STEER-") + ? "STEER-ACK-REPLY" + : `REPLY:${prompt.replaceAll("\n", " ").slice(0, 60)}`; + send(chunk({ content: text }, null)); + send(chunk({}, "stop")); + } + send("[DONE]"); + controller.close(); + }, + }); + return new Response(stream, { + headers: { "content-type": "text/event-stream" }, + }); + }, + }); + return { mock, requests }; +} + +// Conversation requests only: a turn's asynchronous topic-label generation +// can outlive it and is not run activity. +const conversationCount = (requests: { topic: boolean }[]) => + requests.filter((request) => !request.topic).length; + +type SteerManager = ReturnType; + +function bootSteerOffice(room: string, profile: string, mockPort: number) { + const events: unknown[] = []; + const supervisor = new OpenCodeSupervisor({ + profileDir: `${STATE_ROOT}/${profile}`, + serverCwd: STATE_ROOT, + launchEnv: openCodeLiveLaunchEnv(`${STATE_ROOT}/${profile}`), + // Long enough that the server stays warm between a test's turns. + idleShutdownMs: 30_000, + config: openCodeLiveConfig("gate/gate-model", { + gate: openCodeMockProvider(`http://127.0.0.1:${mockPort}/v1`), + }), + }); + reaper.track(supervisor); + const backend = createOpenCodeBackend({ supervisor }); + const sink: EventHandler = (event) => { + events.push(event); + }; + const mgr = createAgentManager({ + eventSink: sink, + resolveBackend: () => backend, + officeState: new OfficeState({ rooms: rooms(room) }), + initialRooms: [], + }); + mgr.configureAgentTurnDeps(); + return { mgr, supervisor, events }; +} + +async function waitUntil( + pred: () => boolean | Promise, + label: string, + timeoutMs = 30_000, + onTimeout?: () => unknown, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (await pred()) return; + if (Date.now() > deadline) { + if (onTimeout) console.info(`live steer timeout diagnostics:`, JSON.stringify(onTimeout(), null, 1)); + throw new Error(`live steer: timed out waiting for ${label}`); + } + await Bun.sleep(20); + } +} + +function settled( + mgr: SteerManager, + agentId: string, + extra: () => boolean = () => true, +): boolean { + const queue = mgr.getAllAgents().find((a) => a.id === agentId)?.queue ?? []; + return ( + mgr.getAgent(agentId)?.state === "waiting_for_response" && + queue.length === 0 && + extra() + ); +} + +function steerEntries(mgr: SteerManager, agentId: string, text: string) { + return mgr + .getAgentLogs(agentId) + .filter( + (entry) => + entry.kind === "user_message" && + typeof entry.content === "string" && + entry.content.includes(text), + ); +} + +// Answers the shell permission with "allow once" (choice 1: OpenCode offers +// no persistent label, so allow_once is first). +async function allowNextPermission( + mgr: SteerManager, + events: unknown[], + agentId: string, + answered: Set, +): Promise { + await waitUntil( + () => + events.some( + (event) => + (event as { type?: string }).type === "interaction_added" && + (event as { interaction?: { kind?: string; agentId?: string } }) + .interaction?.kind === "permission" && + (event as { interaction?: { agentId?: string } }).interaction + ?.agentId === agentId && + !answered.has( + (event as { interaction: { id: string } }).interaction.id, + ), + ), + "permission interaction", + 60_000, + () => ({ + eventTypes: [...new Set(events.map((e) => (e as { type?: string }).type))], + state: mgr.getAgent(agentId)?.state, + logs: mgr + .getAgentLogs(agentId) + .map((entry) => `${entry.kind}: ${String(entry.content).slice(0, 100)}`), + }), + ); + const event = events.find( + (candidate) => + (candidate as { type?: string }).type === "interaction_added" && + (candidate as { interaction?: { kind?: string; agentId?: string } }) + .interaction?.kind === "permission" && + (candidate as { interaction?: { agentId?: string } }).interaction + ?.agentId === agentId && + !answered.has((candidate as { interaction: { id: string } }).interaction.id), + ) as { interaction: { id: string } }; + answered.add(event.interaction.id); + const result = mgr.respondToChoiceInteraction( + agentId, + event.interaction.id, + "1", + "tester", + ); + expect(result.ok).toBe(true); +} + +// The session's pending inbox rows, read straight from the real server. +async function liveInbox( + supervisor: OpenCodeSupervisor, + sessionId: string, +): Promise { + const lease = await supervisor.acquire(); + try { + const response = await fetch( + `${lease.baseUrl}/api/session/${encodeURIComponent(sessionId)}/inbox`, + { headers: { authorization: lease.authHeader } }, + ); + const body = (await response.json()) as { data?: unknown[] }; + return body.data ?? []; + } finally { + lease.release(); + } +} + +async function spawnReceiver( + mgr: SteerManager, + room: string, +): Promise { + const info = await mgr.spawn( + "Steer Receiver", + STATE_ROOT, + "default", + undefined, + undefined, + room, + undefined, + "gate/gate-model", + "high", + undefined, + "opencode", + ); + if (!info) throw new Error("live steer: spawn returned null"); + return info.id; +} + +it.skipIf(!LIVE)( + "steer during a long shell call folds into the running turn on the real server", + async () => { + const { mock, requests } = steerMock({}); + const { mgr, supervisor, events } = bootSteerOffice( + "room-live-steer-c1", + "opencode-steer-c1-profile", + mock.port ?? 0, + ); + const teardown = reaper.add(async () => { + await supervisor.shutdown(); + await mock.stop(true); + }); + try { + const agentId = await spawnReceiver(mgr, "room-live-steer-c1"); + const answered = new Set(); + mgr.enqueueMessage(agentId, { + sender: { kind: "user", username: "tester" }, + text: "Please run LONG-SHELL now.", + }); + await allowNextPermission(mgr, events, agentId, answered); + // The tool now runs its sleep 4: the run has no boundary until it ends. + await waitUntil( + () => + ["thinking", "tool_executing"].includes( + mgr.getAgent(agentId)?.state ?? "", + ), + "tool running", + 60_000, + ); + const steerText = "STEER-C1 FYI from Sender while the shell runs."; + const ack = mgr.enqueueMessage( + agentId, + { sender: STEER_SENDER, text: steerText }, + { steer: true }, + ); + expect(ack).toMatchObject({ ok: true, queued: false, steered: true }); + await waitUntil( + () => settled(mgr, agentId), + "turn settled", + 60_000, + ); + + // Logged exactly once, as a tool-boundary delivery (steer_delivered), + // before the turn's final reply. + const entries = steerEntries(mgr, agentId, "STEER-C1"); + expect(entries).toHaveLength(1); + expect(entries[0].metadata?.delivery).toBe("tool_boundary"); + const logs = mgr.getAgentLogs(agentId); + expect(logs.indexOf(entries[0]!)).toBeLessThan( + logs.findIndex((entry) => entry.content?.includes("STEER-ACK-REPLY")), + ); + // The tool ran to its end (the sleep was not cut) and the next model + // request carried its result plus the steer text: OpenCode promoted + // the row at the boundary, inside the same run. Keyed on the last + // user message, since the topic-label request also holds the history. + const lastUserOf = (request: { messages: Array> }) => { + const last = [...request.messages] + .reverse() + .find((message) => message.role === "user")?.content; + return typeof last === "string" ? last : JSON.stringify(last ?? ""); + }; + const postTool = requests.find( + (request) => + lastUserOf(request).includes("STEER-C1") && + request.messages.some((message) => message.role === "tool"), + ); + expect(postTool).toBeDefined(); + // The marker in the tool role's own content is the sleep's stdout (the + // command text also names it, in the call's arguments). + expect( + postTool!.messages.some( + (message) => + message.role === "tool" && + JSON.stringify(message.content ?? "").includes(SLEPT_TO_END), + ), + ).toBe(true); + // The inbox is drained and nothing runs on its own after the turn. + const sessionId = mgr.getCurrentSessionId(agentId); + expect(sessionId).toBeTruthy(); + expect(await liveInbox(supervisor, sessionId!)).toEqual([]); + const before = conversationCount(requests); + await Bun.sleep(2_000); + expect(conversationCount(requests)).toBe(before); + } finally { + await teardown(); + } + }, + 90_000, +); + +it.skipIf(!LIVE)( + "a steer that misses the run is cancelled and the flush delivers it exactly once", + async () => { + const mockState: SteerMockState = {}; + const { mock, requests } = steerMock(mockState); + const { mgr, supervisor } = bootSteerOffice( + "room-live-steer-c2", + "opencode-steer-c2-profile", + mock.port ?? 0, + ); + const teardown = reaper.add(async () => { + await supervisor.shutdown(); + await mock.stop(true); + }); + try { + const agentId = await spawnReceiver(mgr, "room-live-steer-c2"); + const classifications: string[] = []; + // Swept delays after the mock started streaming the run's final + // answer: the miss window (admit after the run's last boundary check, + // before the turn's end cancels) is milliseconds wide, so attempts + // land folded, missed, or after the flip to idle. Every outcome must + // deliver exactly once; the test requires at least one real miss. + const offsets = [0, 3, 6, 10, 15, 25, 40, 70]; + for ( + let attempt = 0; + attempt < offsets.length && !classifications.includes("miss"); + attempt++ + ) { + const steerText = `STEER-C2-${attempt} note from Sender.`; + let signal!: () => void; + const finalStarted = new Promise((resolve) => { + signal = resolve; + }); + mockState.onFinal = signal; + mgr.enqueueMessage(agentId, { + sender: { kind: "user", username: "tester" }, + text: `FAST-ANSWER-${attempt} please reply.`, + }); + await Promise.race([ + finalStarted, + Bun.sleep(30_000).then(() => { + throw new Error("live steer: final answer never streamed"); + }), + ]); + await Bun.sleep(offsets[attempt]!); + const ack = mgr.enqueueMessage( + agentId, + { sender: STEER_SENDER, text: steerText }, + { steer: true }, + ); + expect(ack.ok).toBe(true); + await waitUntil( + () => settled(mgr, agentId, () => steerEntries(mgr, agentId, steerText).length > 0), + `attempt ${attempt} settled`, + 60_000, + ); + // Delivered exactly once, whichever path took it. + const entries = steerEntries(mgr, agentId, steerText); + expect(entries).toHaveLength(1); + if ((ack as { steered?: boolean }).steered !== true) + classifications.push("idle-late"); + else if (entries[0].metadata?.delivery === "tool_boundary") + classifications.push("folded"); + else classifications.push("miss"); + // The inbox is drained and no unowned run starts after the turn. + expect( + await liveInbox(supervisor, mgr.getCurrentSessionId(agentId)!), + ).toEqual([]); + const before = conversationCount(requests); + await Bun.sleep(1_200); + expect(conversationCount(requests)).toBe(before); + } + console.info( + "live steer race classifications", + JSON.stringify(classifications), + ); + expect(classifications).toContain("miss"); + } finally { + await teardown(); + } + }, + 240_000, +); + +it.skipIf(!LIVE)( + "an abort with a parked steer cancels its row and the flush delivers it once", + async () => { + const { mock, requests } = steerMock({}); + const { mgr, supervisor, events } = bootSteerOffice( + "room-live-steer-c3", + "opencode-steer-c3-profile", + mock.port ?? 0, + ); + const teardown = reaper.add(async () => { + await supervisor.shutdown(); + await mock.stop(true); + }); + try { + const agentId = await spawnReceiver(mgr, "room-live-steer-c3"); + const answered = new Set(); + mgr.enqueueMessage(agentId, { + sender: { kind: "user", username: "tester" }, + text: "Please run LONG-SHELL now.", + }); + await allowNextPermission(mgr, events, agentId, answered); + await waitUntil( + () => + ["thinking", "tool_executing"].includes( + mgr.getAgent(agentId)?.state ?? "", + ), + "tool running", + 60_000, + ); + const steerText = "STEER-C3 correction from Sender."; + const ack = mgr.enqueueMessage( + agentId, + { sender: STEER_SENDER, text: steerText }, + { steer: true }, + ); + expect(ack).toMatchObject({ ok: true, queued: false, steered: true }); + // The admit lands while the sleep still runs (it is durable in the + // inbox, and the run has no boundary until the tool ends). + await Bun.sleep(500); + const result = await mgr.abort(agentId); + expect(result.ok).toBe(true); + await waitUntil( + () => + settled(mgr, agentId, () => steerEntries(mgr, agentId, steerText).length > 0), + "abort settled with the flush delivered", + 60_000, + ); + + // The member's stop cut the tool: the sleep never reached its echo. + // Checked on tool-role content only - the command text itself names + // the marker in the call's arguments, which the history keeps. + expect( + requests.some((request) => + request.messages.some( + (message) => + message.role === "tool" && + JSON.stringify(message.content ?? "").includes(SLEPT_TO_END), + ), + ), + ).toBe(false); + expect( + mgr + .getAgentLogs(agentId) + .some( + (entry) => + entry.kind === "tool_result" && + entry.content?.includes(SLEPT_TO_END), + ), + ).toBe(false); + // The parked steer was released and the flush delivered it exactly + // once, as an ordinary queued item (not a tool-boundary delivery). + const entries = steerEntries(mgr, agentId, steerText); + expect(entries).toHaveLength(1); + expect(entries[0].metadata?.delivery).toBeUndefined(); + // The row was cancelled: the inbox is empty and the flush's turn is + // the only request carrying the text, exactly once in its messages + // (a row the cancel missed would ride along as a promoted duplicate). + const sessionId = mgr.getCurrentSessionId(agentId); + expect(sessionId).toBeTruthy(); + expect(await liveInbox(supervisor, sessionId!)).toEqual([]); + const carrying = requests.filter((request) => + JSON.stringify( + request.messages.filter( + (message) => message.role === "user", + ), + ).includes("STEER-C3"), + ); + expect(carrying).toHaveLength(1); + const occurrences = carrying[0]!.messages.filter((message) => + JSON.stringify(message.content ?? "").includes("STEER-C3"), + ).length; + expect(occurrences).toBe(1); + const before = conversationCount(requests); + await Bun.sleep(2_000); + expect(conversationCount(requests)).toBe(before); + } finally { + await teardown(); + } + }, + 90_000, +); From 8b32df4bfe2100fc8c323c980b408dfdf8b171dc Mon Sep 17 00:00:00 2001 From: Erik Costlow Date: Wed, 11 Nov 2026 14:08:34 -0500 Subject: [PATCH 3/4] Format: prettier on the OpenCode steer changes Co-Authored-By: Claude Opus 5.5 --- server/backends/opencode/adapter.test.ts | 7 +-- server/backends/opencode/transport.test.ts | 36 ++++++++----- server/backends/opencode/transport.ts | 6 +-- .../agent-manager.opencode.live.test.ts | 54 ++++++++++++------- 4 files changed, 63 insertions(+), 40 deletions(-) diff --git a/server/backends/opencode/adapter.test.ts b/server/backends/opencode/adapter.test.ts index 5fd368ce6..b8e0c8507 100644 --- a/server/backends/opencode/adapter.test.ts +++ b/server/backends/opencode/adapter.test.ts @@ -1601,9 +1601,10 @@ describe("OpenCode session steer", () => { Date.now() < askAt ) await Bun.sleep(2); - const ask = events.find( - (e) => e.kind === "approval_request", - ) as Extract; + const ask = events.find((e) => e.kind === "approval_request") as Extract< + NormalizedEvent, + { kind: "approval_request" } + >; expect(ask).toBeDefined(); expect(await session.steer!("FYI from Sender", "steer-1")).toBe(true); await session.approve(ask.approvalId, { kind: "allow_once" }); diff --git a/server/backends/opencode/transport.test.ts b/server/backends/opencode/transport.test.ts index a1784e378..a3151f2c3 100644 --- a/server/backends/opencode/transport.test.ts +++ b/server/backends/opencode/transport.test.ts @@ -1831,9 +1831,10 @@ describe("OpenCode 2 native steer", () => { // to its boundary, where 2.2.5 promotes the row. await until(() => h.events.some((e) => e.kind === "approval_request")); expect(await h.transport.steer("FYI from Sender", "steer-a")).toBe(true); - const ask = h.events.find( - (e) => e.kind === "approval_request", - ) as Extract; + const ask = h.events.find((e) => e.kind === "approval_request") as Extract< + NormalizedEvent, + { kind: "approval_request" } + >; await h.transport.approve(ask.approvalId, { kind: "allow_once" }); const { events, end } = await sending; @@ -1859,7 +1860,9 @@ describe("OpenCode 2 native steer", () => { // The boundary delivered it: one steer_delivered, before the completion, // and nothing for the foreign row. No cancel was owed. const delivered = events.filter((e) => e.kind === "steer_delivered"); - expect(delivered).toEqual([{ kind: "steer_delivered", steerId: "steer-a" }]); + expect(delivered).toEqual([ + { kind: "steer_delivered", steerId: "steer-a" }, + ]); expect(events.indexOf(delivered[0]!)).toBeLessThan( events.findIndex((e) => e.kind === "turn_completed"), ); @@ -1888,9 +1891,10 @@ describe("OpenCode 2 native steer", () => { const sending = h.turn([{ type: "text", text: "go" }]); await until(() => h.events.some((e) => e.kind === "approval_request")); expect(await h.transport.steer("refused", "steer-r")).toBe(false); - const ask = h.events.find( - (e) => e.kind === "approval_request", - ) as Extract; + const ask = h.events.find((e) => e.kind === "approval_request") as Extract< + NormalizedEvent, + { kind: "approval_request" } + >; await h.transport.approve(ask.approvalId, { kind: "allow_once" }); const { events, end } = await sending; expect(end?.status).toBe("completed"); @@ -1913,9 +1917,10 @@ describe("OpenCode 2 native steer", () => { }); await until(() => h.events.some((e) => e.kind === "approval_request")); expect(await h.transport.steer("too late", "steer-p")).toBe(true); - const ask = h.events.find( - (e) => e.kind === "approval_request", - ) as Extract; + const ask = h.events.find((e) => e.kind === "approval_request") as Extract< + NormalizedEvent, + { kind: "approval_request" } + >; await h.transport.approve(ask.approvalId, { kind: "allow_once" }); const { events, end } = await sending; @@ -1935,9 +1940,10 @@ describe("OpenCode 2 native steer", () => { const sending = h.turn([{ type: "text", text: "go" }]); await until(() => h.events.some((e) => e.kind === "approval_request")); expect(await h.transport.steer("quietly taken", "steer-s")).toBe(true); - const ask = h.events.find( - (e) => e.kind === "approval_request", - ) as Extract; + const ask = h.events.find((e) => e.kind === "approval_request") as Extract< + NormalizedEvent, + { kind: "approval_request" } + >; await h.transport.approve(ask.approvalId, { kind: "allow_once" }); const { events, end } = await sending; @@ -1950,7 +1956,9 @@ describe("OpenCode 2 native steer", () => { // The promotion means the model read the steer: the delivery is reported // before the completion, exactly once, so the flush does not resend it. const delivered = events.filter((e) => e.kind === "steer_delivered"); - expect(delivered).toEqual([{ kind: "steer_delivered", steerId: "steer-s" }]); + expect(delivered).toEqual([ + { kind: "steer_delivered", steerId: "steer-s" }, + ]); expect(events.indexOf(delivered[0]!)).toBeLessThan( events.findIndex((e) => e.kind === "turn_completed"), ); diff --git a/server/backends/opencode/transport.ts b/server/backends/opencode/transport.ts index 7d4614fc5..55537b687 100644 --- a/server/backends/opencode/transport.ts +++ b/server/backends/opencode/transport.ts @@ -1322,9 +1322,9 @@ export class OpenCodeTransport { let pending = false; try { const list = listData( - await this - .request(`/api/session/${session}/inbox`) - .then((r) => r.json()), + await this.request(`/api/session/${session}/inbox`).then((r) => + r.json(), + ), ); pending = list.some( (row) => stringField(asRecord(row), "id") === inboxId, diff --git a/server/test-support/agent-manager.opencode.live.test.ts b/server/test-support/agent-manager.opencode.live.test.ts index ac5e97cbb..91bd72573 100644 --- a/server/test-support/agent-manager.opencode.live.test.ts +++ b/server/test-support/agent-manager.opencode.live.test.ts @@ -447,7 +447,9 @@ function steerMock(state: SteerMockState) { .reverse() .find((message) => message.role === "user")?.content; const prompt = - typeof lastUser === "string" ? lastUser : JSON.stringify(lastUser ?? ""); + typeof lastUser === "string" + ? lastUser + : JSON.stringify(lastUser ?? ""); requests.push({ messages, topic: prompt.includes("topic label") }); const hasToolResult = messages.some((message) => message.role === "tool"); const id = `gate-${requests.length}`; @@ -551,7 +553,11 @@ async function waitUntil( for (;;) { if (await pred()) return; if (Date.now() > deadline) { - if (onTimeout) console.info(`live steer timeout diagnostics:`, JSON.stringify(onTimeout(), null, 1)); + if (onTimeout) + console.info( + `live steer timeout diagnostics:`, + JSON.stringify(onTimeout(), null, 1), + ); throw new Error(`live steer: timed out waiting for ${label}`); } await Bun.sleep(20); @@ -606,11 +612,15 @@ async function allowNextPermission( "permission interaction", 60_000, () => ({ - eventTypes: [...new Set(events.map((e) => (e as { type?: string }).type))], + eventTypes: [ + ...new Set(events.map((e) => (e as { type?: string }).type)), + ], state: mgr.getAgent(agentId)?.state, logs: mgr .getAgentLogs(agentId) - .map((entry) => `${entry.kind}: ${String(entry.content).slice(0, 100)}`), + .map( + (entry) => `${entry.kind}: ${String(entry.content).slice(0, 100)}`, + ), }), ); const event = events.find( @@ -620,7 +630,9 @@ async function allowNextPermission( .interaction?.kind === "permission" && (candidate as { interaction?: { agentId?: string } }).interaction ?.agentId === agentId && - !answered.has((candidate as { interaction: { id: string } }).interaction.id), + !answered.has( + (candidate as { interaction: { id: string } }).interaction.id, + ), ) as { interaction: { id: string } }; answered.add(event.interaction.id); const result = mgr.respondToChoiceInteraction( @@ -650,10 +662,7 @@ async function liveInbox( } } -async function spawnReceiver( - mgr: SteerManager, - room: string, -): Promise { +async function spawnReceiver(mgr: SteerManager, room: string): Promise { const info = await mgr.spawn( "Steer Receiver", STATE_ROOT, @@ -708,11 +717,7 @@ it.skipIf(!LIVE)( { steer: true }, ); expect(ack).toMatchObject({ ok: true, queued: false, steered: true }); - await waitUntil( - () => settled(mgr, agentId), - "turn settled", - 60_000, - ); + await waitUntil(() => settled(mgr, agentId), "turn settled", 60_000); // Logged exactly once, as a tool-boundary delivery (steer_delivered), // before the turn's final reply. @@ -727,7 +732,9 @@ it.skipIf(!LIVE)( // request carried its result plus the steer text: OpenCode promoted // the row at the boundary, inside the same run. Keyed on the last // user message, since the topic-label request also holds the history. - const lastUserOf = (request: { messages: Array> }) => { + const lastUserOf = (request: { + messages: Array>; + }) => { const last = [...request.messages] .reverse() .find((message) => message.role === "user")?.content; @@ -814,7 +821,12 @@ it.skipIf(!LIVE)( ); expect(ack.ok).toBe(true); await waitUntil( - () => settled(mgr, agentId, () => steerEntries(mgr, agentId, steerText).length > 0), + () => + settled( + mgr, + agentId, + () => steerEntries(mgr, agentId, steerText).length > 0, + ), `attempt ${attempt} settled`, 60_000, ); @@ -889,7 +901,11 @@ it.skipIf(!LIVE)( expect(result.ok).toBe(true); await waitUntil( () => - settled(mgr, agentId, () => steerEntries(mgr, agentId, steerText).length > 0), + settled( + mgr, + agentId, + () => steerEntries(mgr, agentId, steerText).length > 0, + ), "abort settled with the flush delivered", 60_000, ); @@ -928,9 +944,7 @@ it.skipIf(!LIVE)( expect(await liveInbox(supervisor, sessionId!)).toEqual([]); const carrying = requests.filter((request) => JSON.stringify( - request.messages.filter( - (message) => message.role === "user", - ), + request.messages.filter((message) => message.role === "user"), ).includes("STEER-C3"), ); expect(carrying).toHaveLength(1); From 46d16eefceb3a87188efa6b2297b3be9a208b1e2 Mon Sep 17 00:00:00 2001 From: Erik Costlow Date: Wed, 11 Nov 2026 18:41:58 -0500 Subject: [PATCH 4/4] =?UTF-8?q?Docs:=20steer=20=C2=A73.5=20names=20which?= =?UTF-8?q?=20half=20of=20the=20OpenCode=20inbox=20contract=20upstream=20t?= =?UTF-8?q?ests,=20and=20which=20our=20live=20tests=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- internal-docs/steer-delivery-design.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/internal-docs/steer-delivery-design.md b/internal-docs/steer-delivery-design.md index b3297f00a..a41379c6b 100644 --- a/internal-docs/steer-delivery-design.md +++ b/internal-docs/steer-delivery-design.md @@ -264,6 +264,13 @@ against a live server, since a row it misses is promoted by the next run before **Edit and fork.** As on Codex, the steer is a backend user message inside the turn; its text starts with the note and the sender prefix, and `editMessage` already excludes `tool_boundary` entries (§3.1). +**Upstream guarantee.** OpenCode's own suite pins only half of this: "only records the prompt when resume is +false" (`packages/core/test/session-prompt.test.ts`) checks that `resume: false` never wakes the session. A test +for the other half, a steer row that stays pending and cancellable, was proposed and declined +([anomalyco/opencode#63356](https://github.com/anomalyco/opencode/pull/63356), closed as not planned). Only the +live tests in `server/test-support/agent-manager.opencode.live.test.ts` guard it, so run +`bun run test:opencode` on every OpenCode version bump. + ## 4. What changes for callers | Surface | Before | After |