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.