From 402e813bd042cd4f0643ac4d2490666588261287 Mon Sep 17 00:00:00 2001 From: OmarAlex24 Date: Sun, 15 Nov 2026 19:51:38 +0000 Subject: [PATCH] fix(sandbox): a sandbox chat's own wake runs reach the host's chat When the sandbox's thread woke itself between host turns (a background task or subagent finishing, a monitor line, a scheduled task), the reply stayed on the sandbox: the bridge only sent the host its background roster, so the phone and any client not paired with the sandbox saw the chat go idle without the answer. The bridge now keeps following the sandbox's thread between turns and, for each run the thread starts on a message no user typed, asks the host for a run of its own (SANDBOX_BRIDGE_WAKE_METHOD). The host's ACP adapter offers that as a provider continuation carrying the sandbox's message id, text and notification, so the mirror gets a wake run like any T3 thread. That run's prompt names the sandbox's message, which the bridge follows with the relay normal turns use; a provider continuation's text is never sent to the sandbox. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpRegistryAdapterV2.test.ts | 98 +++++++++++ docs/internals/sandbox-environments.md | 17 +- docs/user/remote-access.md | 5 +- packages/contracts/src/sandbox.ts | 6 + .../src/server/adapter.ts | 16 +- packages/provider-acp/src/server/adapter.ts | 38 ++++- packages/sandbox-bridge/src/bridge.test.ts | 154 ++++++++++++++++++ packages/sandbox-bridge/src/bridge.ts | 107 +++++++++++- 8 files changed, 424 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index ae912f52f5b6e..e45a574c3a214 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -14,7 +14,9 @@ import { SANDBOX_BRIDGE_BACKGROUND_WORK_METHOD, SANDBOX_BRIDGE_CANCEL_META, SANDBOX_BRIDGE_INSTANCE_ID, + SANDBOX_BRIDGE_MESSAGE_META, SANDBOX_BRIDGE_STEER_CANCEL, + SANDBOX_BRIDGE_WAKE_METHOD, ThreadId, } from "@t3tools/contracts"; import { pendingBackgroundTurnItems } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; @@ -1427,6 +1429,102 @@ describe("AcpRegistryAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect("starts a run when the sandbox bridge reports one the sandbox began itself", () => + Effect.gen(function* () { + const notification = { + source: { kind: "subagent" }, + outcome: "completed", + summary: "Review the diff finished", + } as const; + const wake = { + type: "emit_inbound", + frame: { + kind: "notification", + method: SANDBOX_BRIDGE_WAKE_METHOD, + params: { + sessionId: CLAUDE_ROOT_SESSION, + messageId: "sandbox-wake-1", + text: "Background task completed.", + notification, + }, + }, + }; + const gate = makeProviderReplayGate(["wake"]); + const session = yield* openClaudeReplaySession( + [ + ...claudeFrames.open, + { + type: "expect_outbound", + frame: { kind: "request", method: "session/set_config_option", params: "" }, + }, + { + type: "emit_inbound", + frame: { + kind: "response", + method: "session/set_config_option", + result: { configOptions: [] }, + }, + }, + ...claudeFrames.prompt(1), + claudeFrames.idle("end_turn"), + // Repeated while one is pending, it asks for nothing more. + { ...wake, label: "wake" }, + wake, + claudeMarker("woke"), + // The bridge follows the sandbox's run by the message id its prompt carries. + { + type: "expect_outbound", + frame: { + kind: "request", + method: "session/prompt", + params: { + sessionId: CLAUDE_ROOT_SESSION, + prompt: "", + _meta: { + [SANDBOX_BRIDGE_MESSAGE_META]: { + messageId: "sandbox-wake-1", + createdBy: "agent", + creationSource: "provider", + }, + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { kind: "response", method: "session/prompt", result: {} }, + }, + claudeFrames.idle("end_turn"), + claudeFrames.closeRequest, + claudeFrames.closed, + ], + gate, + true, + "fixture-agent", + SANDBOX_BRIDGE_INSTANCE_ID, + ); + yield* session.runtime.startTurn(session.turnInput(1)); + yield* session.terminal; + gate.release("wake"); + yield* Queue.take(session.offered); + yield* session.throughMarker("woke"); + assert.lengthOf(session.offers, 1); + const [offer] = session.offers; + assert.strictEqual(offer?.threadId, session.turnInput(1).threadId); + assert.strictEqual(offer?.providerThreadId, session.providerThread.id); + assert.strictEqual(offer?.messageId, "sandbox-wake-1"); + assert.strictEqual(offer?.detail, "Background task completed."); + assert.deepEqual(offer?.notification, notification); + // The server dispatches the request as the host thread's message, whose run reaches here. + const turn = session.turnInput(2, { continuation: true }); + yield* session.runtime.startTurn({ + ...turn, + message: { ...turn.message, messageId: MessageId.make("sandbox-wake-1") }, + }); + assert.strictEqual((yield* session.terminal).status, "completed"); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("gives Copilot T3's MCP server over HTTP when it advertises HTTP", () => Effect.gen(function* () { const selfInvocation = yield* resolveSelfInvocation(); diff --git a/docs/internals/sandbox-environments.md b/docs/internals/sandbox-environments.md index a1d8974f3b90c..2d33e47b04297 100644 --- a/docs/internals/sandbox-environments.md +++ b/docs/internals/sandbox-environments.md @@ -430,14 +430,25 @@ without pairing with the sandbox. in any chat, also stops the thread's delegated tasks), else `run.interrupt`. - **Work after the turn.** A host turn ends with the sandbox run carrying its message, but the sandbox thread can keep working: subagents, background tasks, runs it wakes itself for. So the - bridge keeps following it between turns and sends what it still works on (its live runs, and - the pending work its own shell would hold for, by the shared pending-work rules) as - `SANDBOX_BRIDGE_BACKGROUND_WORK_METHOD`, then an empty list once it settles. The host's ACP adapter + bridge keeps following it until the next prompt or the end of its process, not only until it + settles, since a task's end and the wake it causes arrive apart. It sends what the thread still + works on (its live runs, and the pending work its own shell would hold for, by the shared + pending-work rules) as `SANDBOX_BRIDGE_BACKGROUND_WORK_METHOD`, an empty list once it settles. The host's ACP adapter (`reportsBackgroundTasks`) makes that the mirror's provider-thread roster, so the mirror's shell holds it Working on every client, the phone included, with no per-row subscription and no pairing with the sandbox. The roster belongs to the bridge process: a holding one keeps the process from idle release, and it is cleared when the process ends. Stop on the held mirror reaches the bridge as a plain cancel between turns, which stops the sandbox's thread. +- **Wakes.** A run the sandbox's thread starts itself (on a message no user typed: a wake, a + delegated result, a scheduled task, its own continue) after the newest run a host turn followed goes to the host as + `SANDBOX_BRIDGE_WAKE_METHOD`, with the sandbox's message id, text and notification (without the + notification's sandbox ids). The host's ACP adapter offers it as a provider continuation + (`requestContinuation`), one pending at a time and none while a turn runs, so the mirror gets a + wake run as any T3 thread does, on every client. That run's prompt carries the sandbox's message + id, so the bridge follows it as a message the sandbox already has, and never sends a provider + continuation's text to the sandbox. Each wake is a host run of its own: once one ends, the + bridge follows again and asks for the next. A send during it steers or queues as on any running + thread, and a Stop stops the sandbox's thread. - **Credential.** After the launch hand-off the provisioner runs [`bridgeCredentialScript`](../../apps/server/src/sandbox/sandboxBridgeCredential.ts) on the sandbox: `t3 auth session issue` with `orchestration:read`, `orchestration:operate` and diff --git a/docs/user/remote-access.md b/docs/user/remote-access.md index c3aa71016b111..a415ba8e02a8c 100644 --- a/docs/user/remote-access.md +++ b/docs/user/remote-access.md @@ -150,8 +150,9 @@ Connections → Sandboxes** has **Connect** and **Disconnect** for each chat's s behind it. Once the sandbox is running and connected, the chat shows the sandbox's own thread again. - Attachments and images are not forwarded yet. -- The thread under **Sandboxes** shows only what was sent through it; messages sent to the sandbox - directly appear on the sandbox's own thread. +- The thread under **Sandboxes** shows what was sent through it, and the replies the sandbox's + agent gives on its own, such as when a background task or subagent finishes; messages sent to + the sandbox directly appear on the sandbox's own thread. - The **Sandboxes** provider shows only in projects that hold a sandbox thread. ### Settle or delete a sandbox chat diff --git a/packages/contracts/src/sandbox.ts b/packages/contracts/src/sandbox.ts index aed07dcdf1f3e..76c4f663071de 100644 --- a/packages/contracts/src/sandbox.ts +++ b/packages/contracts/src/sandbox.ts @@ -1151,6 +1151,12 @@ export const SANDBOX_BRIDGE_WITHDRAW_METHOD = "_t3code/sandbox_requests_withdraw * ended, as the host thread's background roster; an empty list once it settles. */ export const SANDBOX_BRIDGE_BACKGROUND_WORK_METHOD = "_t3code/sandbox_background_work"; +/** + * The bridge's ACP notification that the sandbox's thread started a run on its own between host + * turns: the host starts a provider continuation with the sandbox's message id, which the bridge + * then follows like any message the sandbox already has. + */ +export const SANDBOX_BRIDGE_WAKE_METHOD = "_t3code/sandbox_wake"; /** * Provider instances T3 seeds and runs for itself, like the sandbox bridge. They work wherever a diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index f021aa262b5ca..9d1a52eb5a73c 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -20,12 +20,14 @@ import { import { defaultInstanceIdForDriver, ProviderDriverKind, + OrchestrationV2Notification, OrchestrationV2PendingBackgroundTask, SANDBOX_BRIDGE_BACKGROUND_WORK_METHOD, SANDBOX_BRIDGE_CANCEL_META, SANDBOX_BRIDGE_INSTANCE_ID, SANDBOX_BRIDGE_MESSAGE_META, SANDBOX_BRIDGE_STEER_CANCEL, + SANDBOX_BRIDGE_WAKE_METHOD, SANDBOX_BRIDGE_WITHDRAW_METHOD, } from "@t3tools/contracts"; import { readCustomModelEntries } from "@t3tools/shared/model"; @@ -131,10 +133,17 @@ const SandboxBackgroundWork = Schema.Struct({ sessionId: Schema.String, tasks: Schema.Array(OrchestrationV2PendingBackgroundTask), }); +const SandboxWake = Schema.Struct({ + sessionId: Schema.String, + messageId: Schema.String, + text: Schema.String, + notification: Schema.optional(OrchestrationV2Notification), +}); /** * The sandbox bridge withdraws a request once it is answered on the sandbox, so both sides close - * it, and reports what the sandbox's thread still works on after a turn as the thread's roster. + * it, reports what the sandbox's thread still works on after a turn as the thread's roster, and + * asks for a run of the host thread's own when the sandbox's thread starts one by itself. */ function registerSandboxBridgeExtensions(context: AcpAdapterV2ExtensionContext) { return Effect.all( @@ -149,6 +158,11 @@ function registerSandboxBridgeExtensions(context: AcpAdapterV2ExtensionContext) SandboxBackgroundWork, context.reportBackgroundTasks, ), + context.runtime.handleExtNotification( + SANDBOX_BRIDGE_WAKE_METHOD, + SandboxWake, + context.requestContinuation, + ), ], { discard: true }, ); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index b1a6ba9756ed0..2984b7eb02958 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -2,9 +2,11 @@ import { type ChatAttachment, + MessageId, type ModelSelection, type OrchestrationV2ConversationMessage, type OrchestrationV2ExecutionNode, + type OrchestrationV2Notification, type OrchestrationV2PlanArtifact, type OrchestrationV2PlanStep, type OrchestrationV2ProviderCapabilities, @@ -280,6 +282,20 @@ export interface AcpAdapterV2ExtensionContext { readonly sessionId: string; readonly tasks: ReadonlyArray; }) => Effect.Effect; + /** + * Asks for a provider continuation between turns, for an agent that starts work on its own. Its + * prompt carries the message, so the agent can tell which of its own runs to show there. + */ + readonly requestContinuation: ( + notice: { readonly sessionId: string } & AcpAdapterV2ContinuationMessage, + ) => Effect.Effect; +} + +/** The message a requested continuation starts its run with. */ +export interface AcpAdapterV2ContinuationMessage { + readonly messageId: string; + readonly text: string; + readonly notification?: OrchestrationV2Notification | undefined; } /** @@ -4723,7 +4739,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }); }); - const offerContinuationRun = Effect.fnUntraced(function* (_sessionId: string) { + const offerContinuationRun = Effect.fnUntraced(function* ( + _sessionId: string, + message?: AcpAdapterV2ContinuationMessage, + ) { if (continuationRequests === undefined) { return false; } @@ -4748,7 +4767,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); if (Option.isNone(pending)) return false; const { route, generation, reports } = pending.value; - const notification = backgroundWorkNotification([...reports.values()]); + const notification = + message?.notification ?? backgroundWorkNotification([...reports.values()]); yield* Effect.logInfo("orchestration-v2.acp-wake-turn-detected", { driver, providerSessionId: input.providerSessionId, @@ -4759,7 +4779,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( threadId: route.threadId, providerThreadId: route.providerThreadId, driver, - detail: null, + detail: message?.text ?? null, + ...(message === undefined ? {} : { messageId: MessageId.make(message.messageId) }), ...(notification === null ? {} : { notification }), clearIfCurrent: () => continuationPermit.withPermit( @@ -7070,6 +7091,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }); }), ).pipe(Effect.asVoid), + requestContinuation: ({ sessionId, ...message }) => + runRuntimeCallbackAtGeneration( + handlerGeneration, + Effect.gen(function* () { + if ((yield* Ref.get(activeSessionId)) !== sessionId) return; + // A notice that crossed the next prompt belongs to that turn, not a new run. + const context = yield* Ref.get(activeTurn); + if (context !== null && !context.finalized) return; + yield* offerContinuationRun(sessionId, message); + }), + ).pipe(Effect.asVoid), applyBackgroundTaskMutation: (incoming) => runRuntimeCallbackAtGeneration( handlerGeneration, diff --git a/packages/sandbox-bridge/src/bridge.test.ts b/packages/sandbox-bridge/src/bridge.test.ts index 56bdbc5e34241..c63c11a5f0bb3 100644 --- a/packages/sandbox-bridge/src/bridge.test.ts +++ b/packages/sandbox-bridge/src/bridge.test.ts @@ -5,6 +5,7 @@ import { SANDBOX_BRIDGE_CANCEL_META, SANDBOX_BRIDGE_MESSAGE_META, SANDBOX_BRIDGE_STEER_CANCEL, + SANDBOX_BRIDGE_WAKE_METHOD, SANDBOX_BRIDGE_WITHDRAW_METHOD, SANDBOX_RESUME_STEP_SECONDS, type OrchestrationV2Command, @@ -361,6 +362,8 @@ interface Harness { readonly withdrawals: Array; /** The background rosters the bridge reported, in order. */ readonly rosters: Queue.Queue; + /** The runs the bridge asked the host for, in order. */ + readonly wakes: Queue.Queue; } const messages = ( @@ -435,12 +438,16 @@ const withBridge = ( elicitations: [], withdrawals: [], rosters: yield* Queue.unbounded(), + wakes: yield* Queue.unbounded(), }; yield* acp.handleExtNotification( SANDBOX_BRIDGE_BACKGROUND_WORK_METHOD, Schema.Unknown, (payload) => Queue.offer(harness.rosters, payload).pipe(Effect.asVoid), ); + yield* acp.handleExtNotification(SANDBOX_BRIDGE_WAKE_METHOD, Schema.Unknown, (payload) => + Queue.offer(harness.wakes, payload).pipe(Effect.asVoid), + ); yield* acp.handleExtNotification(SANDBOX_BRIDGE_WITHDRAW_METHOD, Schema.Unknown, (payload) => Effect.sync(() => void harness.withdrawals.push(payload)), ); @@ -1672,6 +1679,153 @@ describe("forwarding a turn", () => { }), ); + it.effect("shows a run the sandbox starts itself as a run of the host's, once", () => + Effect.gen(function* () { + const reviewer = (status: string) => + ({ + id: "item-reviewer", + type: "subagent", + title: "Review the diff", + status, + ordinal: 3, + runId: "run-1", + childThreadId: "sandbox-child", + updatedAt: at(), + }) as never; + // The reviewer's result, which the sandbox's thread gives itself once the reviewer ends. + const wakeMessage = { + id: "wake-1", + role: "user", + createdBy: "agent", + creationSource: "provider", + text: "Background task completed.", + notification: { + source: { kind: "subagent", childThreadId: "sandbox-child" }, + outcome: "completed", + summary: "Review the diff finished", + }, + } as never; + const wakeRun = (status: string) => run("wake-1" as MessageId, status, "run-2", 2); + const wakeItems = (reply: string) => [ + reviewer("completed"), + userMessage("wake-1", 300, "run-2", 2_000), + assistant("wake-answer", reply, 301, "run-2", 2_001), + ]; + const fake = yield* makeFakeSandbox({ + existing: projection([]), + react: (command, sandbox) => { + if (command.type !== "message.dispatch") return Effect.void; + const ran = run(command.messageId!, "completed"); + sandbox.settle(projection([reviewer("running")], [ran])); + return Effect.forEach( + [ + event("run.created", run(command.messageId!, "running")), + event("turn-item.updated", reviewer("running")), + event("turn-item.updated", assistant("answer", "Started a reviewer", 2)), + event("run.updated", ran), + ], + sandbox.push, + { discard: true }, + ); + }, + }); + const subscribed = yield* Queue.unbounded(); + const counted: SandboxSession = { + ...fake.session, + subscribe: (id) => + Stream.unwrap( + Queue.offer(subscribed, undefined).pipe(Effect.as(fake.session.subscribe(id))), + ), + }; + yield* withBridge( + { fake, connect: () => Effect.succeed(counted) }, + ({ acp, updates, rosters, wakes }) => + Effect.gen(function* () { + const session = yield* openThread(acp); + const roster = (tasks: ReadonlyArray) => ({ + sessionId: session.sessionId, + tasks, + }); + yield* say(acp, session.sessionId, "Review it in the background", { + messageId: "msg-1", + }); + assert.deepStrictEqual( + yield* Queue.take(rosters), + roster([ + { taskId: "item-reviewer", kind: "subagent", description: "Review the diff" }, + ]), + ); + // The reviewer ends before the run it wakes the sandbox's thread for begins. + yield* fake.push(event("turn-item.updated", reviewer("completed"))); + assert.deepStrictEqual(yield* Queue.take(rosters), roster([])); + yield* fake.push(event("message.updated", wakeMessage)); + yield* fake.push(event("run.created", wakeRun("running"))); + // The host is asked for a run with the sandbox's message, its notice in host terms. + assert.deepStrictEqual(yield* Queue.take(wakes), { + sessionId: session.sessionId, + messageId: "wake-1", + text: "Background task completed.", + notification: { + source: { kind: "subagent" }, + outcome: "completed", + summary: "Review the diff finished", + }, + }); + assert.deepStrictEqual( + yield* Queue.take(rosters), + roster([ + { + taskId: "sandbox-run:run-2", + kind: "background_task", + description: "Working on the sandbox", + }, + ]), + ); + + // That run's prompt follows the sandbox's run: what it said so far, then the rest. + const before = updates.length; + fake.settle( + projection(wakeItems("Looks"), [ + run("msg-1" as MessageId, "completed"), + wakeRun("running"), + ]), + ); + const wake = yield* say(acp, session.sessionId, "Background task completed.", { + messageId: "wake-1", + createdBy: "agent", + creationSource: "provider", + }).pipe(Effect.forkScoped); + // The first turn, the follow after it, and this one. + yield* Effect.forEach([1, 2, 3], () => Queue.take(subscribed), { discard: true }); + fake.settle( + projection(wakeItems("Looks good"), [ + run("msg-1" as MessageId, "completed"), + wakeRun("completed"), + ]), + ); + yield* fake.push( + event( + "turn-item.updated", + assistant("wake-answer", "Looks good", 301, "run-2", 2_002), + ), + ); + yield* fake.push(event("run.updated", wakeRun("completed"))); + assert.strictEqual((yield* Fiber.join(wake)).stopReason, "end_turn"); + assert.deepStrictEqual(messages(updates.slice(before)), ["Looks", " good"]); + assert.deepStrictEqual( + fake.commands.flatMap((command) => + command.type === "message.dispatch" ? [String(command.messageId)] : [], + ), + ["msg-1"], + ); + // Following again after it, the bridge does not ask for that run a second time. + assert.deepStrictEqual(yield* Queue.take(rosters), roster([])); + assert.strictEqual(yield* Queue.size(wakes), 0); + }), + ); + }), + ); + it.effect("waits out a slow resume and then sends the message it was given", () => Effect.gen(function* () { const fake = yield* makeFakeSandbox({ react: answerWith("Back up") }); diff --git a/packages/sandbox-bridge/src/bridge.ts b/packages/sandbox-bridge/src/bridge.ts index ddfc006a5b0ca..223f7b3c417ff 100644 --- a/packages/sandbox-bridge/src/bridge.ts +++ b/packages/sandbox-bridge/src/bridge.ts @@ -12,6 +12,7 @@ import { type ModelSelection, type OrchestrationV2Command, type OrchestrationV2ConversationMessage, + type OrchestrationV2Notification, type OrchestrationV2PendingBackgroundTask, type OrchestrationV2ProviderThread, type OrchestrationV2Run, @@ -33,6 +34,7 @@ import { SANDBOX_BRIDGE_MESSAGE_META, SANDBOX_BRIDGE_MODEL, SANDBOX_BRIDGE_STEER_CANCEL, + SANDBOX_BRIDGE_WAKE_METHOD, SANDBOX_BRIDGE_WITHDRAW_METHOD, SANDBOX_RESUME_WAIT_SECONDS, type SecretRequestAnswerInput, @@ -1056,6 +1058,25 @@ const hostTask = ( ...(awaited || task.awaited === true ? { awaited: true } : {}), }) as OrchestrationV2PendingBackgroundTask; +// A notice's ids (tasks, child threads, pull requests) are the sandbox's; the host keeps its words. +const hostNotification = (notification: OrchestrationV2Notification): OrchestrationV2Notification => + ({ + ...notification, + source: { + kind: + notification.source.kind === "delegated_task" + ? "background_task" + : notification.source.kind, + }, + }) as OrchestrationV2Notification; + +/** A run the sandbox's thread started itself, which the host shows as a run of its own. */ +export interface SandboxWake { + readonly messageId: MessageId; + readonly text: string; + readonly notification?: OrchestrationV2Notification; +} + /** * What a sandbox thread still works on once the host's turn ended, as the host thread's roster: * runs the sandbox started itself (a wake, a queued message) and the background work its thread @@ -1067,6 +1088,8 @@ export class SandboxWork { private readonly items = new Map(); private readonly providerThreads = new Map(); private readonly wakeMessages = new Set(); + // Messages nobody typed: wakes, delegated results, scheduled tasks, the sandbox's own continue. + private readonly ownPrompts = new Map(); private activeProviderThreadId: string | null = null; /** Whether the item can change the roster. */ @@ -1140,6 +1163,41 @@ export class SandboxWork { ]; } + /** The newest run over when following began; it and every run before it are history. */ + get settledOrdinal(): number { + let newest = 0; + for (const run of this.runs.values()) { + if (isTerminalRun(run)) newest = Math.max(newest, run.ordinal); + } + return newest; + } + + /** Runs after `ordinal` the sandbox's thread started itself, oldest first, once each began. */ + wakesAfter(ordinal: number): ReadonlyArray { + return [...this.runs.values()] + .filter( + (run) => + run.ordinal > ordinal && + run.status !== "queued" && + // Cancelled while it waited in the queue: it never ran, so there is nothing to show. + !(isTerminalRun(run) && run.startedAt === null), + ) + .toSorted((left, right) => left.ordinal - right.ordinal) + .flatMap((run) => { + const message = this.ownPrompts.get(run.userMessageId); + if (message === undefined) return []; + return [ + { + messageId: message.id, + text: message.text, + ...(message.notification === undefined + ? {} + : { notification: hostNotification(message.notification) }), + }, + ]; + }); + } + private keepItem(item: OrchestrationV2TurnItem): boolean { if (!BACKGROUND_ITEM_TYPES.has(item.type)) return false; if (isOrchestrationV2WorkActive(item.status)) this.items.set(item.id, item); @@ -1148,6 +1206,15 @@ export class SandboxWork { } private noteMessage(message: OrchestrationV2ConversationMessage): boolean { + const own = + message.role === "user" && + (message.createdBy !== "user" || message.scheduledTaskId !== undefined) && + !this.ownPrompts.has(message.id); + if (own) this.ownPrompts.set(message.id, message); + return this.noteWakeMessage(message) || own; + } + + private noteWakeMessage(message: OrchestrationV2ConversationMessage): boolean { const notification = message.notification; if ( message.delegatedCompletion !== undefined || @@ -1204,8 +1271,11 @@ export const turnTargetOf = ( const queued = Option.isSome(before) ? before.value.runs.some((run) => run.userMessageId === message.messageId) : false; - return queued - ? { _tag: "attach", messageId: message.messageId, replay: false } + if (queued) return { _tag: "attach", messageId: message.messageId, replay: false }; + // The host's run for one the sandbox began itself (`SANDBOX_BRIDGE_WAKE_METHOD`): its notice + // only names that run, so it is never sent back. + return message.createdBy === "agent" && message.creationSource === "provider" + ? { _tag: "nothing" } : { _tag: "send", messageId: message.messageId }; }; /** The MCP credential the host put in a session's `mcpServers`, over stdio or HTTP. */ @@ -1264,6 +1334,8 @@ interface BridgeSession { link?: SandboxBridgeLink | undefined; /** Follows the sandbox thread's own work between turns; see `followSandboxWork`. */ watch?: Fiber.Fiber; + /** The newest sandbox run a host turn followed: a run it starts itself after that is a wake. */ + followedOrdinal?: number; } type ThreadSnapshot = Extract; @@ -1431,6 +1503,9 @@ export const runSandboxBridge = Effect.fn("SandboxBridge.run")(function* ( mcp, cancel: existing?.cancel ?? null, configOptions, + ...(existing?.followedOrdinal === undefined + ? {} + : { followedOrdinal: existing.followedOrdinal }), }); return { configOptions } satisfies AcpSchema.ResumeSessionResponse; }); @@ -1492,12 +1567,13 @@ export const runSandboxBridge = Effect.fn("SandboxBridge.run")(function* ( ); /** - * After a turn, follows the sandbox thread while it still works (its subagents, background - * tasks, the runs it wakes itself for) and keeps the host thread's roster to that work, so every - * client of the host shows the chat Working until the sandbox's thread settles. + * Between turns, keeps the host thread's roster to the sandbox thread's work, and asks the host + * for a run of its own for each run that thread starts itself, which the next prompt follows. + * It follows past a settle, since a task's end and the wake it causes arrive apart. */ const followSandboxWork = Effect.fn("SandboxBridge.followSandboxWork")(function* ( sessionId: string, + session: BridgeSession, threadId: ThreadId, link: SandboxBridgeLink, ) { @@ -1508,19 +1584,29 @@ export const runSandboxBridge = Effect.fn("SandboxBridge.run")(function* ( const work = new SandboxWork(); // The host's adapter drops a roster it already has, so the first one always goes. let sent: string | null = null; + // Each wake is asked for once here; the host keeps one pending, and after its run this + // follow starts again and asks for the next. + const asked = new Set(); yield* Effect.gen(function* () { const sandbox = yield* reachSandbox(link); yield* sandbox.subscribe(threadId).pipe( - Stream.runForEachWhile((item) => + Stream.runForEach((item) => Effect.gen(function* () { - if (!work.apply(item)) return true; + if (!work.apply(item)) return; + session.followedOrdinal ??= work.settledOrdinal; + for (const wake of work.wakesAfter(session.followedOrdinal)) { + if (asked.has(wake.messageId)) continue; + asked.add(wake.messageId); + yield* agent.client + .extNotification(SANDBOX_BRIDGE_WAKE_METHOD, { sessionId, ...wake }) + .pipe(Effect.ignore); + } const tasks = work.tasks; const key = rosterKey(tasks); if (key !== sent) { sent = key; yield* report(tasks); } - return tasks.length > 0; }), ), ); @@ -1566,6 +1652,7 @@ export const runSandboxBridge = Effect.fn("SandboxBridge.run")(function* ( if (session.threadId !== null && session.link !== undefined) { session.watch = yield* followSandboxWork( request.sessionId, + session, session.threadId, session.link, ).pipe(Effect.forkIn(bridgeScope)); @@ -2112,6 +2199,10 @@ export const runSandboxBridge = Effect.fn("SandboxBridge.run")(function* ( Effect.as("followed" as const), Effect.raceFirst(Deferred.await(cancel)), ); + const followed = tracker.run; + if (followed !== undefined) { + session.followedOrdinal = Math.max(session.followedOrdinal ?? 0, followed.ordinal); + } // What a cancelled turn held back still shows, with the copies it already made. if (outcome !== "followed") yield* flushText(undefined, false); // A steer's cancel leaves the run and its tasks working; the steer follows as the next prompt.