diff --git a/apps/server/src/mcp/WorktreeMcpService.test.ts b/apps/server/src/mcp/WorktreeMcpService.test.ts index 90cf3c7f8401a..cecc066b849d0 100644 --- a/apps/server/src/mcp/WorktreeMcpService.test.ts +++ b/apps/server/src/mcp/WorktreeMcpService.test.ts @@ -4,7 +4,10 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { CommandId, EnvironmentId, + EventId, + RunId, type OrchestrationV2ThreadProjection, + type OrchestrationV2ServerCommand, type Project, ProjectId, ProviderInstanceId, @@ -12,6 +15,7 @@ import { WorktreeMcpHandoffInput, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -28,6 +32,7 @@ import * as ServerConfig from "../config.ts"; import { OrchestratorDispatchError, OrchestratorProjectionError, + type OrchestratorV2DispatchResult, } from "../orchestration-v2/Orchestrator.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../project/ProjectService.ts"; @@ -106,7 +111,7 @@ interface HarnessOptions { readonly threadAttachedOnRecheck?: boolean; readonly threadArchivedOnRecheck?: boolean; readonly threadReadFailsOnRecheck?: boolean; - readonly continuation?: "queued" | "fails" | "dies"; + readonly continuation?: "queued" | "started"; readonly projectMissing?: boolean; readonly projectReadFails?: boolean; readonly existingBranchWorktreePath?: string | null; @@ -122,7 +127,7 @@ interface HarnessOptions { const makeHarness = (options: HarnessOptions = {}) => { const thread = options.thread === undefined ? {} : options.thread; const scope = makeScope(options.capabilities ?? new Set(["preview", "worktree"])); - const dispatch = vi.fn((_: unknown) => + const dispatch = vi.fn((command: OrchestrationV2ServerCommand) => (options.dispatchGate ?? Effect.void).pipe( Effect.andThen( options.dispatchInterrupts @@ -131,7 +136,45 @@ const makeHarness = (options: HarnessOptions = {}) => { ? Effect.die(new Error("dispatch defect")) : options.dispatchFails ? (Effect.fail("simulated dispatch failure") as never) - : Effect.succeed({ sequence: 1, storedEvents: [] }), + : Effect.succeed({ + sequence: 1, + storedEvents: + command.type !== "thread.metadata.update" || + command.worktreeContinuation === undefined + ? [] + : [ + { + sequence: 1, + commandId: command.commandId, + event: { + id: EventId.make("event-worktree-continuation"), + type: "run.created", + threadId, + occurredAt: DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"), + payload: { + id: RunId.make("run-worktree-continuation"), + threadId, + ordinal: 2, + providerInstanceId: ProviderInstanceId.make("provider-test"), + modelSelection: { + instanceId: ProviderInstanceId.make("provider-test"), + model: "test-model", + }, + providerThreadId: null, + userMessageId: command.worktreeContinuation.messageId, + rootNodeId: null, + activeAttemptId: null, + status: options.continuation === "started" ? "starting" : "queued", + requestedAt: DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"), + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }, + }, + ], + } satisfies OrchestratorV2DispatchResult), ), ), ); @@ -179,22 +222,9 @@ const makeHarness = (options: HarnessOptions = {}) => { ? Effect.succeed(makeProjection(thread)) : Effect.fail(new OrchestratorProjectionError({ threadId: id })); }); - const sendToThread = vi.fn((_: unknown) => { - switch (options.continuation ?? "queued") { - case "fails": - return Effect.fail( - new ThreadManagementService.ThreadManagementThreadArchivedError({ - threadId, - }), - ); - case "dies": - return Effect.die(new Error("send defect")); - default: - return Effect.succeed({ - delivery: "queued", - } as ThreadManagementService.ThreadManagementSendResult); - } - }); + const sendToThread = vi.fn(() => + Effect.die(new Error("Standalone continuation dispatch is unsafe")), + ); const getById = vi.fn((id: ProjectId) => options.projectReadFails ? (Effect.fail("simulated project read failure") as never) @@ -398,16 +428,44 @@ const runStatus = (harness: ReturnType) => const decodeHandoffInput = Schema.decodeUnknownEffect(WorktreeMcpHandoffInput); describe("t3_worktree_handoff", () => { - it.effect("reports a continuation failure without failing the handoff", () => { - const harness = makeHarness({ continuation: "fails" }); + it.effect("queues the continuation prompt as the thread's next message", () => { + const harness = makeHarness(); return Effect.gen(function* () { const result = yield* runHandoff(harness, { - branch: "feature/continue-fails", + branch: "feature/continue", + continuationPrompt: "Keep fixing the login bug in the new worktree.", + }); + + expect(result.continuation).toEqual({ status: "scheduled", delivery: "queued" }); + expect(harness.dispatch).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + type: "thread.metadata.update", + threadId, + worktreePath: result.worktreePath, + expectedWorktreePath: null, + worktreeContinuation: { + messageId: expect.stringContaining("worktree-continuation"), + text: "Keep fixing the login bug in the new worktree.", + }, + }), + ); + expect(harness.sendToThread).not.toHaveBeenCalled(); + expect(harness.dispatch.mock.invocationCallOrder[0]).toBeLessThan( + harness.runForThread.mock.invocationCallOrder[0]!, + ); + }); + }); + + it.effect("reports a continuation started when no run was blocking the handoff", () => { + const harness = makeHarness({ continuation: "started" }); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { + branch: "feature/continue-idle", continuationPrompt: "Keep going.", }); - expect(result.continuation).toMatchObject({ status: "failed" }); - expect(result.worktreePath).toBe("/worktrees/project/feature/continue-fails"); - expect(harness.dispatch).toHaveBeenCalled(); + expect(result.continuation).toEqual({ status: "scheduled", delivery: "started" }); + expect(harness.dispatch).toHaveBeenCalledTimes(1); + expect(harness.sendToThread).not.toHaveBeenCalled(); }); }); @@ -637,7 +695,12 @@ describe("t3_worktree_handoff", () => { yield* Fiber.join(interruption); expect(harness.dispatch).toHaveBeenCalledTimes(1); - expect(harness.sendToThread).toHaveBeenCalledTimes(1); + expect(harness.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ + worktreeContinuation: expect.objectContaining({ text: "Keep going in the worktree." }), + }), + ); + expect(harness.sendToThread).not.toHaveBeenCalled(); expect(harness.runForThread).toHaveBeenCalledTimes(1); }), ); diff --git a/apps/server/src/mcp/WorktreeMcpService.ts b/apps/server/src/mcp/WorktreeMcpService.ts index 0ee45bb7f8bb3..a514c8d1c2cf2 100644 --- a/apps/server/src/mcp/WorktreeMcpService.ts +++ b/apps/server/src/mcp/WorktreeMcpService.ts @@ -147,7 +147,6 @@ const make = Effect.gen(function* () { ); return { commandId: CommandId.make(part("command", "worktree-handoff")), - continuationCommandId: CommandId.make(part("command", "worktree-continuation")), continuationMessageId: MessageId.make(part("message", "worktree-continuation")), }; }), @@ -293,7 +292,7 @@ const make = Effect.gen(function* () { `Thread '${scope.thread.threadId}' was archived while the worktree was being created; the handoff was rolled back.`, ); } - yield* threadManagement + return yield* threadManagement .dispatch({ type: "thread.metadata.update", commandId: ids.commandId, @@ -301,6 +300,14 @@ const make = Effect.gen(function* () { branch: worktree.worktree.refName, worktreePath, expectedWorktreePath: null, + ...(input.continuationPrompt === undefined + ? {} + : { + worktreeContinuation: { + messageId: ids.continuationMessageId, + text: input.continuationPrompt, + }, + }), }) .pipe( Effect.catchCause((cause) => @@ -321,33 +328,14 @@ const make = Effect.gen(function* () { ), ); - // Queue continuation immediately after binding; detachment terminates the caller, so relaunch needs a durable message. - const queueContinuation: Effect.Effect = - Effect.suspend(() => - input.continuationPrompt === undefined - ? Effect.succeed({ status: "skipped" }) - : threadManagement - .sendToThread({ - projectId: projection.thread.projectId, - commandId: ids.continuationCommandId, - threadId: scope.thread.threadId, - messageId: ids.continuationMessageId, - text: input.continuationPrompt, - attachments: [], - mode: "queue", - createdBy: "agent", - creationSource: "mcp", - }) - .pipe( - Effect.map((sendResult): WorktreeMcpContinuationStatus => ({ - status: "scheduled", - delivery: sendResult.delivery, - })), - reportFailed("worktree handoff continuation failed to queue"), - ), - ); - - const continuation = yield* recheckAndBind.pipe(Effect.andThen(queueContinuation)); + const bound = yield* recheckAndBind; + const queued = bound.storedEvents.some( + ({ event }) => event.type === "run.created" && event.payload.status === "queued", + ); + const continuation: WorktreeMcpContinuationStatus = + input.continuationPrompt === undefined + ? { status: "skipped" } + : { status: "scheduled", delivery: queued ? "queued" : "started" }; yield* vcsStatusBroadcaster .refreshStatus(worktreePath) diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 92e1e01ad43f2..3fd9c675aa4d6 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -38,6 +38,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ type: Schema.Literal("provider-session.detach"), providerSessionId: ProviderSessionId, detail: Schema.optional(Schema.String), + reason: Schema.optional(Schema.Literal("workspace_changed")), revokeMcpCredential: Schema.optional(Schema.Boolean), }), Schema.Struct({ diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index c5b5ea9b383a9..443e58615fc47 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -124,6 +124,7 @@ export const layerExecutor: Layer.Layer< providerSessionId: effect.request.providerSessionId, threadId: effect.threadId, ...(effect.request.detail === undefined ? {} : { detail: effect.request.detail }), + ...(effect.request.reason === undefined ? {} : { reason: effect.request.reason }), ...(effect.request.revokeMcpCredential === undefined ? {} : { revokeMcpCredential: effect.request.revokeMcpCredential }), diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 07c1a544798be..f3486d8e74e2c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3572,6 +3572,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} branch changed before the metadata update could be applied.`, }); } + if (command.type === "thread.metadata.update" && command.worktreeContinuation !== undefined) { + if (isProviderNativeSubagentThread(thread)) { + return yield* new OrchestratorSubagentThreadReadOnlyError({ + commandId: command.commandId, + threadId: command.threadId, + }); + } + if ( + command.worktreePath == null || + command.worktreePath === thread.worktreePath || + thread.archivedAt !== null + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: + "A worktree continuation requires an active thread and a changed worktree binding.", + }); + } + } if (command.type === "thread.metadata.update" && command.expectedEmpty === true) { const records = yield* projectionStore .getThreadRecords(command.threadId, ["runs"]) @@ -4621,6 +4641,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? "Runtime mode changed." : "Provider or model selection changed.", // Terminal detaches revoke MCP credentials; reattachments need other detach reasons to preserve them. + ...(command.type === "thread.metadata.update" + ? { reason: "workspace_changed" as const } + : {}), ...(command.type === "thread.archive" ? { revokeMcpCredential: true } : {}), }, } satisfies PendingOrchestrationEffectV2; @@ -5850,6 +5873,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command: Extract, events: Ref.Ref>, effects: Ref.Ref>, + worktreeContinuationOfRunId?: RunId, ) { let projection = yield* getProjectionWithPendingEvents(command.threadId, events); if (command.manualContinuationOfRunId !== undefined) { @@ -5921,6 +5945,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( !source || !( + (source.id === worktreeContinuationOfRunId && isActiveRun(source)) || isRestartContinuationSource(source, projection.providerTurns) || continuationFailedBeforePrompt(source, projection.providerTurns) ) || @@ -5936,6 +5961,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) || (yield* stopReachedRun(command, command.threadId, source.id)) ) { + if (worktreeContinuationOfRunId !== undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "The active run changed before the workspace continuation could be queued.", + }); + } // Preserve the current row so stale automatic deliveries receive an // accepted receipt without changing work or repeatedly retrying. yield* emit( @@ -13630,7 +13662,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.active.reorder": case "thread.tag.set": case "thread.mark-unread": - case "thread.metadata.update": case "thread.pull-request.link": case "thread.pull-request.unlink": case "thread.pull-request-link.sync": @@ -13643,6 +13674,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "provider.switch": yield* dispatchThreadMutation(command, events, effects); break; + case "thread.metadata.update": { + const sourceRun = + command.worktreeContinuation === undefined + ? undefined + : (yield* getProjectionWithPendingEvents(command.threadId, events)).runs.find( + isActiveRun, + ); + yield* dispatchThreadMutation(command, events, effects); + if (command.worktreeContinuation !== undefined) { + yield* dispatchMessage( + { + type: "message.dispatch", + commandId: command.commandId, + threadId: command.threadId, + ...command.worktreeContinuation, + ...(sourceRun ? { restartContinuationOfRunId: sourceRun.id } : {}), + attachments: [], + createdBy: "agent", + creationSource: "mcp", + dispatchMode: { type: "queue_after_active" }, + }, + events, + effects, + sourceRun?.id, + ); + } + break; + } case "thread.pull-request-watch.sync": yield* dispatchPullRequestWatchSync(command, events, effects); break; diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 25ae18f7b6690..5464fdef86958 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1702,8 +1702,8 @@ it.effect( modelSelection, runtimePolicy, }); - const firstClient = yield* runtime.subscribeEvents!; - const secondClient = yield* runtime.subscribeEvents!; + const firstClient = yield* runtime.subscribeEvents!(); + const secondClient = yield* runtime.subscribeEvents!(); const oldQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(oldQueue); const statuses = [{ key: "watch", text: "Watching 4 files" }]; @@ -1758,7 +1758,7 @@ it.effect( modelSelection, runtimePolicy, }); - const replacementClient = yield* replacement.subscribeEvents!; + const replacementClient = yield* replacement.subscribeEvents!(); const newQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(newQueue); yield* Queue.offer(oldQueue!, { @@ -1823,7 +1823,7 @@ it.effect.each(["session metadata", "session status changes", "provider work"] a modelSelection, runtimePolicy, }); - const subscriber = yield* runtime.subscribeEvents!; + const subscriber = yield* runtime.subscribeEvents!(); const observed = yield* Queue.unbounded(); yield* subscriber.events.pipe( Stream.runForEach((event) => Queue.offer(observed, event)), @@ -1934,8 +1934,8 @@ it.effect("ProviderSessionManagerV2 interrupts event subscriptions on server shu modelSelection, runtimePolicy, }); - const bufferedSubscription = yield* runtime.subscribeEvents!; - const activeSubscription = yield* runtime.subscribeEvents!; + const bufferedSubscription = yield* runtime.subscribeEvents!(); + const activeSubscription = yield* runtime.subscribeEvents!(); const adapterQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(adapterQueue); yield* Queue.offer(adapterQueue!, { @@ -1977,7 +1977,7 @@ it.effect("ProviderSessionManagerV2 drains subscribers when the provider stops", modelSelection, runtimePolicy, }); - const subscription = yield* runtime.subscribeEvents!; + const subscription = yield* runtime.subscribeEvents!(); const collected = yield* subscription.events.pipe(Stream.runCollect, Effect.forkScoped); const adapterQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(adapterQueue); @@ -3983,8 +3983,8 @@ it.effect( const subscribe = firstRuntime.subscribeEvents; assert.isDefined(subscribe); if (subscribe === undefined) return; - const firstSubscription = yield* subscribe; - const secondSubscription = yield* subscribe; + const firstSubscription = yield* subscribe(); + const secondSubscription = yield* subscribe(); const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(queue); yield* Queue.offer(queue!, { @@ -4252,7 +4252,7 @@ it.effect( const publish = Effect.fnUntraced(function* ( event: ProviderAdapter.ProviderAdapterV2Event, ) { - const subscription = yield* failed.runtime.subscribeEvents!; + const subscription = yield* failed.runtime.subscribeEvents!(); yield* Queue.offer(queue!, event); // Observing the exact event ensures the manager has handled it before resuming. assert.deepEqual(yield* subscription.events.pipe(Stream.runHead), Option.some(event)); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 84a64b10c81f2..11e60d57070a9 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -191,6 +191,7 @@ export interface ProviderSessionManagerV2Shape { readonly providerSessionId: ProviderSessionId; readonly threadId: ThreadId; readonly detail?: string; + readonly reason?: "workspace_changed"; readonly revokeMcpCredential?: boolean; }) => Effect.Effect; /** @@ -234,9 +235,7 @@ interface LiveSessionEntry { readonly supportsMultipleProviderThreads: boolean; readonly runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime; readonly exposedRuntime: ProviderAdapter.ProviderAdapterV2SessionRuntime; - readonly eventSubscribers: Ref.Ref< - ReadonlyMap> - >; + readonly eventSubscribers: Ref.Ref>; readonly requestEventPermit: Semaphore.Semaphore; readonly scope: Scope.Closeable; readonly idleGeneration: number; @@ -252,6 +251,11 @@ interface IdleThreadUnload { readonly fiber: Fiber.Fiber | null; } +interface ProviderSessionEventSubscriber { + readonly threadId: ThreadId | undefined; + readonly queue: Queue.Queue; +} + type ProviderSessionEventSignal = | { readonly type: "event"; readonly event: ProviderAdapter.ProviderAdapterV2Event } | { @@ -553,14 +557,12 @@ export const layerWithOptions = ( ); const publishToSubscribers = ( - subscribers: Ref.Ref< - ReadonlyMap> - >, + subscribers: Ref.Ref>, signal: ProviderSessionEventSignal, ) => Ref.get(subscribers).pipe( Effect.flatMap((current) => - Effect.forEach(current.values(), (queue) => Queue.offer(queue, signal), { + Effect.forEach(current.values(), ({ queue }) => Queue.offer(queue, signal), { discard: true, }), ), @@ -568,17 +570,23 @@ export const layerWithOptions = ( const failSubscribers = Effect.fn("ProviderSessionManager.failSubscribers")(function* ( entry: LiveSessionEntry, - detail: string, + error: ProviderAdapter.ProviderAdapterV2Error, + threadId?: ThreadId, ) { - const error = new ProviderAdapter.ProviderAdapterEventStreamError({ - driver: entry.runtime.driver, - providerSessionId: entry.runtime.providerSessionId, - cause: detail, + const subscribers = yield* Ref.modify(entry.eventSubscribers, (current) => { + const failed: Array = []; + const remaining = new Map(current); + for (const [id, subscriber] of current) { + if (threadId === undefined || subscriber.threadId === threadId) { + failed.push(subscriber); + remaining.delete(id); + } + } + return [failed, remaining] as const; }); - const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); yield* Effect.forEach( - subscribers.values(), - (queue) => + subscribers, + ({ queue }) => Queue.offer(queue, { type: "failure", cause: Cause.fail(error), @@ -595,7 +603,7 @@ export const layerWithOptions = ( const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); yield* Effect.forEach( subscribers.values(), - (queue) => Queue.clear(queue).pipe(Effect.andThen(Queue.interrupt(queue))), + ({ queue }) => Queue.clear(queue).pipe(Effect.andThen(Queue.interrupt(queue))), { discard: true }, ); }, @@ -607,7 +615,7 @@ export const layerWithOptions = ( entry: LiveSessionEntry, ) { const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); - yield* Effect.forEach(subscribers.values(), (queue) => Queue.end(queue), { + yield* Effect.forEach(subscribers.values(), ({ queue }) => Queue.end(queue), { discard: true, }); }); @@ -1103,7 +1111,11 @@ export const layerWithOptions = ( } else { yield* failSubscribers( entry, - input.detail ?? `Provider session released: ${input.reason}.`, + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: entry.runtime.driver, + providerSessionId: entry.runtime.providerSessionId, + cause: input.detail ?? `Provider session released: ${input.reason}.`, + }), ); } const closeExit = yield* closeScopeWithin(entry.scope, { @@ -1894,15 +1906,14 @@ export const layerWithOptions = ( const makeEventSubscription = Effect.fn("ProviderSessionManager.makeEventSubscription")( function* ( - subscribers: Ref.Ref< - ReadonlyMap> - >, + subscribers: Ref.Ref>, + threadId?: ThreadId, ): Effect.fn.Return { const queue = yield* Queue.unbounded(); const subscriberId = yield* Ref.getAndUpdate(nextSubscriberId, (value) => value + 1); yield* Ref.update(subscribers, (current) => { const updated = new Map(current); - updated.set(subscriberId, queue); + updated.set(subscriberId, { threadId, queue }); return updated; }); const close = Ref.modify(subscribers, (current) => { @@ -1933,12 +1944,11 @@ export const layerWithOptions = ( const decorateRuntime = ( runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime, - eventSubscribers: Ref.Ref< - ReadonlyMap> - >, + eventSubscribers: Ref.Ref>, ): ProviderAdapter.ProviderAdapterV2SessionRuntime => { const providerSessionId = runtime.providerSessionId; - const subscribeEvents = makeEventSubscription(eventSubscribers); + const subscribeEvents = (threadId?: ThreadId) => + makeEventSubscription(eventSubscribers, threadId); const turnMetrics = (operation: string, model?: string) => withMetrics({ counter: providerTurnsTotal, @@ -1957,7 +1967,7 @@ export const layerWithOptions = ( }, subscribeEvents, events: Stream.unwrap( - subscribeEvents.pipe(Effect.map((subscription) => subscription.events)), + subscribeEvents().pipe(Effect.map((subscription) => subscription.events)), ), ensureThread: (input) => observeActivity( @@ -2519,7 +2529,7 @@ export const layerWithOptions = ( }), ); const eventSubscribers = yield* Ref.make< - ReadonlyMap> + ReadonlyMap >(new Map()); const exposedRuntime = decorateRuntime(runtime, eventSubscribers); const entry: LiveSessionEntry = { @@ -2689,6 +2699,7 @@ export const layerWithOptions = ( const key = sessionKey(input.providerSessionId); const currentEntry = (yield* Ref.get(sessions)).get(key); let detachedProviderThreads: ReadonlyArray = []; + let interruptActiveTurns = Effect.void; if (currentEntry?.supportsMultipleProviderThreads === true) { const projection = yield* Effect.option( projectionStore.getThreadRecords(input.threadId, [ @@ -2706,7 +2717,7 @@ export const layerWithOptions = ( const activeTurns = projection.value.providerTurns.filter( (turn) => turn.status === "running" && providerThreads.has(turn.providerThreadId), ); - yield* Effect.forEach( + interruptActiveTurns = Effect.forEach( activeTurns, (turn) => currentEntry.exposedRuntime @@ -2789,6 +2800,19 @@ export const layerWithOptions = ( if (Option.isNone(detached)) { return; } + if (input.reason !== undefined) { + yield* failSubscribers( + detached.value, + new ProviderAdapter.ProviderAdapterSessionDetached({ + driver: detached.value.runtime.driver, + providerSessionId: input.providerSessionId, + threadId: input.threadId, + reason: input.reason, + }), + input.threadId, + ); + } + yield* interruptActiveTurns; if ( detached.value.attachedThreadIds.size === 0 && !detached.value.supportsMultipleProviderThreads diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index e25b4c7ab556a..e532e99374cb1 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2266,98 +2266,99 @@ it.effect( providerSessionId: ProviderSessionId.make("session:subagent-interrupt-cascade"), session: { events: Stream.empty, - subscribeEvents: Effect.succeed({ - events: Stream.fromIterable([ - childThreadCreatedEvent(ids), - { - type: "subagent.updated", - driver, - subagent: runningSubagent, - }, - { - type: "node.updated", - driver, - node: runningNode, - }, - { - type: "node.updated", - driver, - node: runningChildNode, - }, - { - type: "node.updated", - driver, - node: unrelatedChildNode, - }, - { - type: "turn_item.updated", - driver, - turnItem: runningTurnItem, - }, - { - type: "turn_item.updated", - driver, - turnItem: runningChildTurnItem, - }, - { - type: "turn_item.updated", - driver, - turnItem: suppressedChildAssistantTurnItem, - }, - { - type: "turn_item.updated", - driver, - turnItem: unrelatedChildTurnItem, - }, - { type: "subagent.updated", driver, subagent: runningMember }, - { type: "subagent.updated", driver, subagent: foreignSubagent }, - { type: "node.updated", driver, node: runningMemberRoot }, - { type: "node.updated", driver, node: foreignChildNode }, - { type: "turn_item.updated", driver, turnItem: foreignChildTurnItem }, - rootTerminalEvent(ids, "interrupted"), - { - type: "subagent.updated", - driver, - subagent: { - ...runningSubagent, - status: "completed" as const, - result: "should-not-apply", - completedAt: runningSubagent.updatedAt, + subscribeEvents: () => + Effect.succeed({ + events: Stream.fromIterable([ + childThreadCreatedEvent(ids), + { + type: "subagent.updated", + driver, + subagent: runningSubagent, }, - }, - { - type: "turn_item.updated", - driver, - turnItem: { - ...runningTurnItem, - status: "completed" as const, - result: "should-not-apply", - completedAt: runningTurnItem.updatedAt, - updatedAt: runningTurnItem.updatedAt, + { + type: "node.updated", + driver, + node: runningNode, }, - }, - { - type: "turn_item.updated", - driver, - turnItem: { - ...runningChildTurnItem, - status: "completed" as const, - completedAt: runningChildTurnItem.updatedAt, - updatedAt: runningChildTurnItem.updatedAt, + { + type: "node.updated", + driver, + node: runningChildNode, }, - }, - { - type: "node.updated", - driver, - node: { - ...runningChildNode, - status: "completed" as const, - completedAt: runningChildNode.startedAt, + { + type: "node.updated", + driver, + node: unrelatedChildNode, }, - }, - ] satisfies ReadonlyArray), - close: Deferred.succeed(ingestionDone, undefined), - }), + { + type: "turn_item.updated", + driver, + turnItem: runningTurnItem, + }, + { + type: "turn_item.updated", + driver, + turnItem: runningChildTurnItem, + }, + { + type: "turn_item.updated", + driver, + turnItem: suppressedChildAssistantTurnItem, + }, + { + type: "turn_item.updated", + driver, + turnItem: unrelatedChildTurnItem, + }, + { type: "subagent.updated", driver, subagent: runningMember }, + { type: "subagent.updated", driver, subagent: foreignSubagent }, + { type: "node.updated", driver, node: runningMemberRoot }, + { type: "node.updated", driver, node: foreignChildNode }, + { type: "turn_item.updated", driver, turnItem: foreignChildTurnItem }, + rootTerminalEvent(ids, "interrupted"), + { + type: "subagent.updated", + driver, + subagent: { + ...runningSubagent, + status: "completed" as const, + result: "should-not-apply", + completedAt: runningSubagent.updatedAt, + }, + }, + { + type: "turn_item.updated", + driver, + turnItem: { + ...runningTurnItem, + status: "completed" as const, + result: "should-not-apply", + completedAt: runningTurnItem.updatedAt, + updatedAt: runningTurnItem.updatedAt, + }, + }, + { + type: "turn_item.updated", + driver, + turnItem: { + ...runningChildTurnItem, + status: "completed" as const, + completedAt: runningChildTurnItem.updatedAt, + updatedAt: runningChildTurnItem.updatedAt, + }, + }, + { + type: "node.updated", + driver, + node: { + ...runningChildNode, + status: "completed" as const, + completedAt: runningChildNode.startedAt, + }, + }, + ] satisfies ReadonlyArray), + close: Deferred.succeed(ingestionDone, undefined), + }), startTurn: () => Effect.void, } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, run: { @@ -2715,6 +2716,127 @@ it.effect("does not refresh pull requests for auxiliary or stale provider termin }), ); +it.effect("interrupts open tool items but preserves completed tools on a planned detach", () => + Effect.gen(function* () { + const key = "workspace-detach-open-tools"; + const ids = backgroundScenarioIds(key); + const runningId = TurnItemId.make(`${ids.itemId}:running`); + const completedId = TurnItemId.make(`${ids.itemId}:completed`); + const { ingested, observed } = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + readAhead: true, + events: (ids) => + Stream.concat( + Stream.fromIterable([ + backgroundTurnItemEvent(ids, "dynamic_tool", "pending", 1), + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 2, runningId), + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 3, completedId), + backgroundTurnItemEvent(ids, "dynamic_tool", "completed", 3, completedId), + ]), + Stream.fail( + new ProviderAdapter.ProviderAdapterSessionDetached({ + driver, + providerSessionId: ProviderSessionId.make(`session:${key}`), + threadId: ids.threadId, + reason: "workspace_changed", + }), + ), + ), + }); + assert.deepEqual( + ingested.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.status === "interrupted" + ? [event.turnItem.id] + : [], + ), + [ids.itemId, runningId], + ); + assert.deepEqual(observed, ["run:interrupted", "pull-requests-refreshed"]); + }), +); + +it.effect("flushes the latest buffered assistant message on a planned detach", () => + Effect.gen(function* () { + const key = "workspace-detach-message"; + const now = yield* DateTime.now; + const { ingested } = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.concat( + Stream.fromIterable( + ["partial", "partial final words"].map((text) => ({ + type: "message.updated" as const, + driver, + message: { + id: MessageId.make(`message:${key}`), + threadId: ids.threadId, + runId: ids.runId, + nodeId: ids.rootNodeId, + role: "assistant" as const, + text, + attachments: [], + streaming: true, + createdBy: "agent" as const, + creationSource: "provider" as const, + createdAt: now, + updatedAt: now, + }, + })), + ), + Stream.fail( + new ProviderAdapter.ProviderAdapterSessionDetached({ + driver, + providerSessionId: ProviderSessionId.make(`session:${key}`), + threadId: ids.threadId, + reason: "workspace_changed", + }), + ), + ), + }); + const messages = ingested.flatMap((event) => + event.type === "message.updated" ? [event.message] : [], + ); + assert.equal(messages.at(-1)?.text, "partial final words"); + assert.equal(messages.at(-1)?.streaming, false); + }), +); + +it.effect( + "ends background items detached after the root has finalized without finalizing twice", + () => + Effect.gen(function* () { + const key = "workspace-detach-after-root"; + const { ingested, observed } = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.concat( + Stream.make( + backgroundTurnItemEvent(ids, "command_execution", "running", 1), + rootTerminalEvent(ids, "completed"), + ), + Stream.fail( + new ProviderAdapter.ProviderAdapterSessionDetached({ + driver, + providerSessionId: ProviderSessionId.make(`session:${key}`), + threadId: ids.threadId, + reason: "workspace_changed", + }), + ), + ), + }); + assert.equal( + ingested.filter( + (event) => event.type === "turn_item.updated" && event.turnItem.status === "interrupted", + ).length, + 1, + ); + assert.deepEqual(observed, ["run:waiting", "pull-requests-refreshed"]); + }), +); + it.effect("refreshes pull requests after a provider stream exits with an error", () => Effect.gen(function* () { const { written, observed } = yield* captureRootRunTermination({ @@ -3206,6 +3328,8 @@ function captureRootRunTermination(input: { readonly afterStart?: ( closed: Deferred.Deferred, ) => Effect.Effect; + // Pull the whole provider stream before processing, as the commit batcher may read ahead. + readonly readAhead?: boolean; }) { return Effect.gen(function* () { const ids = backgroundScenarioIds(input.key); @@ -3219,6 +3343,7 @@ function captureRootRunTermination(input: { }); const writtenItems = yield* Ref.make>([]); const finalEvents = yield* Ref.make>([]); + const ingested = yield* Ref.make>([]); const observed = yield* Ref.make>([]); const submittedEffects = yield* Ref.make>([]); const committedEffects = yield* Ref.make>([]); @@ -3246,6 +3371,17 @@ function captureRootRunTermination(input: { Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, mockEventSink({ + ...(input.readAhead + ? { + consumeProviderEvents: (source, process, after) => + Effect.gen(function* () { + for (const event of yield* Stream.runCollect(source)) { + yield* process(event); + if (yield* after) return; + } + }), + } + : {}), write: (payload) => Effect.gen(function* () { for (const event of payload.events) { @@ -3280,7 +3416,10 @@ function captureRootRunTermination(input: { }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ - ingestNormalized: input.ingestNormalized ?? (() => Effect.succeed([])), + ingestNormalized: + input.ingestNormalized ?? + (({ event }) => + Ref.update(ingested, (current) => [...current, event]).pipe(Effect.as([]))), }), ServerSettings.layerTest(), Layer.succeed(RunFinalizationService.RunFinalizationObserver, { @@ -3303,37 +3442,38 @@ function captureRootRunTermination(input: { providerSessionId: ProviderSessionId.make(`session:${input.key}`), session: { events: Stream.empty, - subscribeEvents: Effect.succeed({ - events: - input.events?.(ids) ?? - Stream.fromIterable([ - ...(input.seedOpenSubagent - ? [ - { type: "subagent.updated", driver, subagent: runningSubagent } as const, - { - type: "node.updated", - driver, - node: makeRunOwnedSubagentNodeFixture({ ids, status: "running" }), - } as const, - { - type: "turn_item.updated", - driver, - turnItem: makeRunOwnedSubagentTurnItemFixture({ - ids, - providerInstanceId, - childThreadId: ids.childThreadId, + subscribeEvents: () => + Effect.succeed({ + events: + input.events?.(ids) ?? + Stream.fromIterable([ + ...(input.seedOpenSubagent + ? [ + { type: "subagent.updated", driver, subagent: runningSubagent } as const, + { + type: "node.updated", driver, - status: "running", - }), - } as const, - ] - : []), - rootTerminalEvent(ids, "interrupted"), - ] satisfies ReadonlyArray), - close: Deferred.succeed(ingestionDone, undefined).pipe( - Effect.andThen(input.onSubscriptionClose ?? Effect.void), - ), - }), + node: makeRunOwnedSubagentNodeFixture({ ids, status: "running" }), + } as const, + { + type: "turn_item.updated", + driver, + turnItem: makeRunOwnedSubagentTurnItemFixture({ + ids, + providerInstanceId, + childThreadId: ids.childThreadId, + driver, + status: "running", + }), + } as const, + ] + : []), + rootTerminalEvent(ids, "interrupted"), + ] satisfies ReadonlyArray), + close: Deferred.succeed(ingestionDone, undefined).pipe( + Effect.andThen(input.onSubscriptionClose ?? Effect.void), + ), + }), startTurn: input.startTurn ?? (() => Effect.void), interruptTurn: input.interruptTurn ?? (() => Effect.void), hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread, @@ -3409,6 +3549,7 @@ function captureRootRunTermination(input: { previewActivityEnded, written: yield* Ref.get(writtenItems), finalEvents: yield* Ref.get(finalEvents), + ingested: yield* Ref.get(ingested), observed: yield* Ref.get(observed), submittedEffects: yield* Ref.get(submittedEffects), committedEffects: yield* Ref.get(committedEffects), @@ -3484,7 +3625,7 @@ function childBackgroundTurnItemEvent( function backgroundTurnItemEvent( ids: BackgroundScenarioIds, type: "command_execution" | "dynamic_tool" | "subagent", - status: "running" | "completed" | "idle", + status: "pending" | "running" | "completed" | "idle", ordinal: number, itemId?: TurnItemId, ): ProviderAdapter.ProviderAdapterV2Event { @@ -3855,27 +3996,28 @@ function runBackgroundItemScenario( ...(options?.hasPendingBackgroundWorkForThread === undefined ? {} : { hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread }), - subscribeEvents: Effect.gen(function* () { - yield* options?.onSubscribe ?? Effect.void; - const source = makeEvents(ids); - const events = (Stream.isStream(source) ? source : Stream.fromIterable(source)).pipe( - Stream.concat( - Stream.fromEffect( - Effect.andThen( - options?.onEventsConsumed ?? Effect.void, - options?.afterEvents?.(previewActivity) ?? Effect.void, - ), - ).pipe(Stream.drain), - ), - ); - return { - events: - options?.keepEventStreamOpen === true - ? events.pipe(Stream.concat(Stream.never)) - : events, - close: Deferred.succeed(ingestionDone, undefined), - }; - }), + subscribeEvents: () => + Effect.gen(function* () { + yield* options?.onSubscribe ?? Effect.void; + const source = makeEvents(ids); + const events = (Stream.isStream(source) ? source : Stream.fromIterable(source)).pipe( + Stream.concat( + Stream.fromEffect( + Effect.andThen( + options?.onEventsConsumed ?? Effect.void, + options?.afterEvents?.(previewActivity) ?? Effect.void, + ), + ).pipe(Stream.drain), + ), + ); + return { + events: + options?.keepEventStreamOpen === true + ? events.pipe(Stream.concat(Stream.never)) + : events, + close: Deferred.succeed(ingestionDone, undefined), + }; + }), startTurn: () => Effect.void, } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, run: { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index e981ab7362d2d..a4e4cb3998744 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -22,6 +22,8 @@ import { type ProviderTurnId, type RunAttemptId, type ThreadId, + type MessageId, + type OrchestrationV2ConversationMessage, type TurnItemId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -642,6 +644,7 @@ export const layer: Layer.Layer< // Root turn snapshot written with the run when T3 ends the run itself. readonly abandonedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; + readonly interruptionMessage?: string; readonly completedAt?: DateTime.Utc; readonly failureItemPersisted: boolean; readonly refreshAfterTurn: Effect.Effect; @@ -810,6 +813,9 @@ export const layer: Layer.Layer< providerThread: input.providerThread, reason, completedAt, + ...(input.interruptionMessage === undefined + ? {} + : { message: input.interruptionMessage }), }), }, ] @@ -1015,6 +1021,9 @@ export const layer: Layer.Layer< const latestTurnItemOrdinal = yield* Ref.make(input.providerTurnOrdinal * 100); const latestProviderThread = yield* Ref.make(input.providerThread); const latestRootProviderTurn = yield* Ref.make(null); + // A planned detach closes ingestion before the adapter can end its open root items. + const openRootTurnItems = new Map(); + const streamingRootMessages = new Map(); // Later root turns in one attempt are provider continuations, such as Codex goal turns. let firstRootProviderTurnId: ProviderTurnId | undefined; const analyticsContext = { @@ -1043,7 +1052,7 @@ export const layer: Layer.Layer< const eventSubscription = input.session.subscribeEvents === undefined ? { events: input.session.events, close: Effect.void } - : yield* input.session.subscribeEvents; + : yield* input.session.subscribeEvents(input.run.threadId); const inheritedBackgroundTurnItems = yield* ( input.loadInheritedBackgroundTurnItems?.() ?? Effect.succeed([]) ).pipe( @@ -1075,6 +1084,7 @@ export const layer: Layer.Layer< ); const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); + const interruptionMessage = yield* Ref.make(undefined); const providerThreadOwnerLost = yield* Ref.make(false); const pendingBackgroundProviderTurns = new Set(); const activeChildProviderTurns = yield* Ref.make>(new Set()); @@ -1097,6 +1107,7 @@ export const layer: Layer.Layer< } const providerThread = yield* Ref.get(latestProviderThread); const openSubagents = yield* Ref.get(openRunOwnedSubagents); + const message = yield* Ref.get(interruptionMessage); yield* writeFinalRunEvents({ run: input.run, rootNode: input.rootNode, @@ -1124,6 +1135,7 @@ export const layer: Layer.Layer< : { loadRunInterruptReason: input.loadRunInterruptReason }), openRunOwnedSubagents: openSubagents, terminal, + ...(message === undefined ? {} : { interruptionMessage: message }), failureItemPersisted: terminal.status === "failed", refreshAfterTurn, }).pipe( @@ -1548,6 +1560,15 @@ export const layer: Layer.Layer< ) { yield* Ref.set(latestRootProviderTurn, event.providerTurn); } + if ( + event.type === "message.updated" && + event.message.threadId === input.run.threadId && + event.message.runId === input.run.id && + event.message.role === "assistant" + ) { + if (event.message.streaming) streamingRootMessages.set(event.message.id, event.message); + else streamingRootMessages.delete(event.message.id); + } let storedEventCount = 0; const deliveredEvent = assistantStreaming.filter( event, @@ -1581,6 +1602,15 @@ export const layer: Layer.Layer< yield* Ref.update(latestTurnItemOrdinal, (current) => Math.max(current, event.turnItem.ordinal), ); + const turnItem = event.turnItem; + // Subagent items are ended by the run-owned cascade. + if (turnItem.type !== "subagent") { + if (isOrchestrationV2WorkActive(turnItem.status)) { + openRootTurnItems.set(turnItem.id, turnItem); + } else { + openRootTurnItems.delete(turnItem.id); + } + } } if (event.type === "turn.terminal") { yield* Ref.set(terminalEvent, event); @@ -1597,17 +1627,92 @@ export const layer: Layer.Layer< yield* trackChildLifecycle(event, deliveredEvent !== null); yield* scheduleIdleFlush; }); + // A planned detach ends the stream. It settles in ingestion order, after buffered events. + const settleDetachedRun = Effect.fnUntraced(function* () { + const rootFinalized = yield* Ref.get(rootRunFinalized); + yield* Ref.set(interruptionMessage, "Run interrupted because the workspace changed"); + const routing = yield* Ref.get(eventRouting); + const providerTurn = yield* Ref.get(latestRootProviderTurn); + const completedAt = yield* DateTime.now; + const terminal = { + type: "turn.terminal", + driver: input.providerThread.driver, + providerThreadId: input.providerThread.id, + providerTurnId: + routing.rootProviderTurnId ?? + input.attempt.providerTurnId ?? + idAllocator.derive.providerTurn({ + driver: input.providerThread.driver, + nativeTurnId: `detached:${input.attempt.id}`, + }), + runOrdinal: input.run.ordinal, + status: "interrupted", + failure: null, + threadDisposition: "reusable", + } satisfies ProviderTerminalEvent; + // Before the provider's first turn event, the detach supplies the root identity for routing. + yield* Ref.update(eventRouting, (state) => + state.rootProviderTurnId === null + ? { + ...state, + rootProviderTurnId: terminal.providerTurnId, + ownedProviderTurns: new Map(state.ownedProviderTurns).set( + terminal.providerTurnId, + input.providerThread.id, + ), + } + : state, + ); + const settlement: ReadonlyArray = [ + ...[...openRootTurnItems.values()].map((turnItem) => ({ + type: "turn_item.updated" as const, + driver: input.providerThread.driver, + turnItem: { + ...turnItem, + ...("streaming" in turnItem ? { streaming: false } : {}), + status: "interrupted" as const, + completedAt, + updatedAt: completedAt, + }, + })), + ...[...streamingRootMessages.values()].map((message) => ({ + type: "message.updated" as const, + driver: input.providerThread.driver, + message: { ...message, streaming: false, updatedAt: completedAt }, + })), + ...(!rootFinalized && + (providerTurn?.status === "running" || providerTurn?.status === "pending") + ? [ + { + type: "provider_turn.updated" as const, + driver: input.providerThread.driver, + providerTurn: { ...providerTurn, status: "interrupted" as const, completedAt }, + }, + ] + : []), + ...(!rootFinalized ? [terminal] : []), + ]; + for (const event of settlement) yield* ingestEvent(yield* routeEvent(event)); + }); const providerEventFiber = yield* eventSink .consumeProviderEvents( eventSubscription.events.pipe( + Stream.catchTag("ProviderAdapterSessionDetached", (detached) => + detached.providerSessionId !== input.providerSessionId || + detached.threadId !== input.run.threadId + ? Stream.fail(detached) + : Stream.succeed({ type: "workspace_detached" as const }), + ), Stream.merge(Stream.fromQueue(idleFlushTicks), { haltStrategy: "left" }), ), (event) => event.type === "idle_flush" ? flushIdleAssistantText(event.at) - : Effect.gen(function* () { - yield* ingestEvent(yield* routeEvent(event)); - }), + : event.type === "workspace_detached" + ? settleDetachedRun() + : Effect.gen(function* () { + yield* ingestEvent(yield* routeEvent(event)); + }), Effect.gen(function* () { if (yield* agentTurnsSettled) yield* endPreviewActivity; return yield* shouldStopProviderEventIngestion; @@ -1905,6 +2010,7 @@ export function makeInterruptResultTurnItem(input: { readonly providerThread: OrchestrationV2ProviderThread; readonly reason: string | undefined; readonly completedAt: DateTime.Utc; + readonly message?: string; }): OrchestrationV2TurnItem { return { id: input.idAllocator.derive.runSignalTurnItem({ @@ -1928,6 +2034,6 @@ export function makeInterruptResultTurnItem(input: { completedAt: input.completedAt, updatedAt: input.completedAt, type: "run_interrupt_result", - message: input.reason ?? "Stopped by the provider", + message: input.reason ?? input.message ?? "Stopped by the provider", }; } diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 4352d8b2d6477..b0300960df154 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -18,6 +18,7 @@ import { ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; @@ -25,10 +26,12 @@ import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EffectWorker from "./EffectWorker.ts"; import * as EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; @@ -86,6 +89,11 @@ function makeRestartAdapter( state: Ref.Ref, sessionCapabilities: OrchestrationV2ProviderCapabilities = pooledCapabilities, providerInstanceId = initialSelection.instanceId, + completeContinuation: Effect.Effect = Effect.void, + beforeTurn?: ( + input: ProviderAdapter.ProviderAdapterV2TurnInput, + publish: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect, + ) => Effect.Effect, ): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: providerInstanceId, @@ -226,11 +234,21 @@ function makeRestartAdapter( }, ], })); + if ( + beforeTurn !== undefined && + (yield* beforeTurn(input, (event) => + Queue.offer(events, event).pipe(Effect.asVoid), + )) + ) + return; const active = { input, providerTurnId: ProviderTurnId.make(`provider-turn:${input.attemptId}`), } satisfies ActiveTurn; - if (input.modelSelection.model === initialSelection.model) { + if ( + input.modelSelection.model === initialSelection.model && + !input.message.text.endsWith("handoff continuation") + ) { const occurredAt = yield* DateTime.now; yield* Ref.update(state, (current) => ({ ...current, activeTurn: active })); yield* Queue.offer(events, { @@ -254,6 +272,48 @@ function makeRestartAdapter( }); return; } + if (input.message.text.endsWith("handoff continuation")) { + const startedAt = yield* DateTime.now; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: active.providerTurnId, + providerThreadId: input.providerThread.id, + nodeId: input.rootNodeId, + runAttemptId: input.attemptId, + nativeTurnRef: { + driver, + nativeId: `native:${active.providerTurnId}`, + strength: "strong", + }, + ordinal: input.providerTurnOrdinal, + status: "running", + startedAt, + completedAt: null, + }, + }); + yield* completeContinuation; + const completedAt = yield* DateTime.now; + yield* Queue.offer(events, { + type: "message.updated", + driver, + message: { + id: MessageId.make(`message:${input.attemptId}:result`), + threadId: input.threadId, + runId: input.runId, + nodeId: input.rootNodeId, + role: "assistant", + text: "Completed inside the new worktree", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "provider", + createdAt: completedAt, + updatedAt: completedAt, + }, + }); + } yield* publishTerminal(active, "completed"); }), steerTurn: () => Effect.void, @@ -897,3 +957,397 @@ it.live.each(["active", "idle", "selection-command", "pooled", "separate-home"] }), ), ); + +it.live("rejects a workspace binding when a provider switch invalidates its continuation", () => + Effect.scoped( + Effect.gen(function* () { + const name = "workspace-continuation-provider-switch"; + const sourceCwd = yield* checkpointWorkspace(`${name}-source`); + const targetCwd = yield* checkpointWorkspace(`${name}-target`); + const threadId = ThreadId.make(name); + const targetId = ProviderInstanceId.make(`${name}-target`); + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: false, + }); + const registry = ProviderAdapterRegistry.layerFromAdapters([ + makeRestartAdapter(state), + makeRestartAdapter(state, pooledCapabilities, targetId), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${name}:create`), + threadId, + projectId: ProjectId.make(`project:${name}`), + title: name, + modelSelection: initialSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: sourceCwd, + createdBy: "user", + creationSource: "web", + }); + const beforeStart = yield* sink.latestSequence({ threadId }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:start`), + threadId, + messageId: MessageId.make(`${name}:start`), + text: "start", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* sink.stream({ threadId, afterSequence: beforeStart }).pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + ); + yield* orchestrator.dispatch({ + type: "provider.switch", + commandId: CommandId.make(`${name}:switch`), + threadId, + modelSelection: { instanceId: targetId, model: initialSelection.model }, + }); + const switched = yield* orchestrator.getThreadProjection(threadId); + assert.equal(switched.thread.providerInstanceId, targetId); + assert.equal(switched.runs[0]?.status, "running"); + // Leave the detach in the outbox: its old run still blocks this handoff. + const failed = yield* orchestrator + .dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`${name}:bind`), + threadId, + worktreePath: targetCwd, + expectedWorktreePath: sourceCwd, + worktreeContinuation: { + messageId: MessageId.make(`${name}:continue`), + text: "handoff continuation", + }, + }) + .pipe(Effect.flip); + assert.include(String(failed.cause), "workspace continuation could be queued"); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.equal(after.thread.worktreePath, sourceCwd); + assert.lengthOf(after.runs, switched.runs.length); + assert.isFalse(after.messages.some((message) => message.id === `${name}:continue`)); + }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, registry))); + }), + ), +); + +it.live.each(["exclusive", "shared"] as const)( + "keeps a delegated %s workspace handoff pending through its atomic continuation", + (sessionMode) => + Effect.scoped( + Effect.gen(function* () { + const name = `delegated-workspace-handoff-${sessionMode}`; + const childReady = yield* Deferred.make<{ + input: ProviderAdapter.ProviderAdapterV2TurnInput; + publish: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; + }>(); + const parentReady = yield* Deferred.make<{ + input: ProviderAdapter.ProviderAdapterV2TurnInput; + publish: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; + }>(); + const sourceCwd = yield* checkpointWorkspace(`${name}-source`); + const targetCwd = yield* checkpointWorkspace(`${name}-target`); + const parentId = ThreadId.make(`thread:${name}:parent`); + const contextReadStarted = yield* Deferred.make(); + const finishContextRead = yield* Deferred.make(); + const continuationStarted = yield* Deferred.make(); + const finishContinuation = yield* Deferred.make(); + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: false, + }); + const registry = ProviderAdapterRegistry.layerSingle( + makeRestartAdapter( + state, + sessionMode === "shared" ? pooledCapabilities : exclusiveCapabilities, + providerInstanceId, + Deferred.succeed(continuationStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishContinuation)), + ), + (input, publish) => + input.message.text.endsWith("bind my workspace") + ? Deferred.succeed(childReady, { input, publish }).pipe(Effect.as(true)) + : Deferred.succeed(parentReady, { input, publish }).pipe(Effect.as(false)), + ), + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${name}:parent:create`), + threadId: parentId, + projectId: ProjectId.make(`project:${name}`), + title: name, + modelSelection: initialSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: sourceCwd, + }); + const beforeStart = yield* sink.latestSequence(); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${name}:parent:start`), + threadId: parentId, + messageId: MessageId.make(`${name}:parent:start`), + text: "delegate the work", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + yield* worker.drain(); + yield* sink.stream({ afterSequence: beforeStart, threadId: parentId }).pipe( + Stream.filter( + (stored) => + stored.event.type === "provider-turn.updated" && + stored.event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + ); + const parentRun = (yield* orchestrator.getThreadProjection(parentId)).runs[0]!; + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make(`${name}:delegate`), + parentThreadId: parentId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "bind my workspace", + modelSelection: initialSelection, + runtimeMode: "full-access", + interactionMode: "default", + }); + const task = (yield* orchestrator.getThreadProjection(parentId)).subagents[0]!; + const childId = task.childThreadId!; + const startingChild = yield* worker.drain(1).pipe(Effect.forkScoped); + yield* Deferred.await(contextReadStarted); + assert.lengthOf((yield* Ref.get(state)).started, 1); + const parent = yield* Deferred.await(parentReady); + const parentSessionId = parent.input.providerThread.providerSessionId; + assert.isNotNull(parentSessionId); + const initialChild = yield* orchestrator.getThreadProjection(childId); + assert.isEmpty(initialChild.providerTurns); + const binding = { + type: "thread.metadata.update" as const, + commandId: CommandId.make(`${name}:bind`), + threadId: childId, + worktreePath: targetCwd, + expectedWorktreePath: sourceCwd, + worktreeContinuation: { + messageId: MessageId.make(`${name}:continue`), + text: "handoff continuation", + }, + }; + const bound = yield* orchestrator.dispatch(binding); + const boundChild = yield* orchestrator.getThreadProjection(childId); + assert.equal(boundChild.thread.worktreePath, targetCwd); + assert.deepEqual( + boundChild.runs.map((run) => run.status), + ["running", "queued"], + ); + assert.equal( + boundChild.messages.find( + (message) => message.id === binding.worktreeContinuation.messageId, + )?.text, + "handoff continuation", + ); + assert.equal( + boundChild.checkpointScopes.find((scope) => scope.runId === boundChild.runs[1]?.id) + ?.cwd, + targetCwd, + ); + assert.deepEqual(yield* orchestrator.dispatch(binding), { ...bound, replayed: true }); + for (const worktreePath of [undefined, targetCwd]) { + const rejected = yield* orchestrator + .dispatch({ + ...binding, + commandId: CommandId.make(`${name}:reject:${worktreePath ?? "branch-only"}`), + branch: "feature/branch-only", + worktreePath, + expectedWorktreePath: targetCwd, + worktreeContinuation: { + messageId: MessageId.make(`${name}:rejected-continue`), + text: "unexpected duplicate continuation", + }, + }) + .pipe(Effect.flip); + assert.include(String(rejected.cause), "changed worktree binding"); + } + // The real outbox must keep the detach behind startup, even while + // the event subscriber exists and the app-context read is suspended. + assert.isFalse(yield* worker.runOnce); + assert.isFalse(yield* Deferred.isDone(childReady)); + yield* Deferred.succeed(finishContextRead, undefined); + yield* Fiber.join(startingChild); + const child = yield* Deferred.await(childReady); + assert.equal( + child.input.providerThread.providerSessionId === + parent.input.providerThread.providerSessionId, + sessionMode === "shared", + ); + yield* worker.drain(1); + // A late provider message is a deterministic fence: a detached subscription + // must end before it can ingest another event for the old run. + const fenceAt = yield* DateTime.now; + const publishFence = ({ input, publish }: typeof child, text: string) => + publish({ + type: "message.updated", + driver, + message: { + id: MessageId.make(`${input.runId}:${text}`), + threadId: input.threadId, + runId: input.runId, + nodeId: input.rootNodeId, + role: "assistant", + text, + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "provider", + createdAt: fenceAt, + updatedAt: fenceAt, + }, + }); + yield* publishFence(child, "late-after-detach"); + yield* EffectWorker.runDaemon.pipe(Effect.forkScoped); + yield* sink.stream({ afterSequence: bound.sequence, threadId: childId }).pipe( + Stream.filter( + ({ event }) => + (event.type === "run.updated" && + event.payload.id === child.input.runId && + ["interrupted", "failed", "completed"].includes(event.payload.status)) || + (event.type === "message.updated" && event.payload.text === "late-after-detach"), + ), + Stream.take(1), + Stream.runDrain, + ); + const detached = yield* orchestrator.getThreadProjection(childId); + assert.equal( + detached.runs[0]?.status, + "interrupted", + "planned detach must terminate the moving run before another provider event is ingested", + ); + assert.equal( + detached.turnItems.find((item) => item.type === "run_interrupt_result")?.message, + "Run interrupted because the workspace changed", + ); + yield* Deferred.await(continuationStarted); + yield* publishFence(parent, "parent-still-alive"); + yield* sink.stream({ afterSequence: bound.sequence, threadId: parentId }).pipe( + Stream.filter( + ({ event }) => + (event.type === "message.updated" && event.payload.text === "parent-still-alive") || + (event.type === "run.updated" && + event.payload.id === parentRun.id && + ["interrupted", "failed", "completed"].includes(event.payload.status)), + ), + Stream.take(1), + Stream.runDrain, + ); + assert.isTrue(yield* manager.isLive(parentSessionId)); + assert.equal( + (yield* orchestrator.getThreadProjection(parentId)).runs[0]?.status, + "running", + ); + const beforeCompletion = yield* orchestrator.getThreadProjection(parentId); + assert.isNull(beforeCompletion.subagents[0]?.result); + assert.isNull(beforeCompletion.subagents[0]?.completionDelivery ?? null); + assert.isFalse( + beforeCompletion.contextTransfers.some( + (transfer) => transfer.type === "subagent_result", + ), + ); + const continuing = yield* orchestrator.getThreadProjection(childId); + assert.deepEqual( + continuing.runs.map((run) => run.status), + ["interrupted", "running"], + ); + assert.isFalse(continuing.runs.some((run) => run.queueHeld)); + assert.equal(continuing.runs[1]?.delegatedTaskId, continuing.runs[0]?.delegatedTaskId); + assert.equal(continuing.runs[1]?.restartContinuationOfRunId, continuing.runs[0]?.id); + const beforeFinish = yield* sink.latestSequence(); + yield* Deferred.succeed(finishContinuation, undefined); + yield* sink.stream({ afterSequence: beforeFinish, threadId: parentId }).pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === task.id && + stored.event.payload.status === "completed", + ), + Stream.take(1), + Stream.runDrain, + ); + const finished = yield* orchestrator.getThreadProjection(parentId); + assert.equal(finished.subagents[0]?.status, "completed"); + assert.equal(finished.subagents[0]?.result, "Completed inside the new worktree"); + assert.lengthOf( + finished.contextTransfers.filter((transfer) => transfer.type === "subagent_result"), + 1, + ); + const childFinished = yield* orchestrator.getThreadProjection(childId); + assert.deepEqual( + childFinished.runs.map((run) => run.status), + ["interrupted", "completed"], + ); + assert.lengthOf( + childFinished.messages.filter((message) => message.text === "handoff continuation"), + 1, + ); + assert.isFalse( + childFinished.messages.some( + (message) => message.text === "unexpected duplicate continuation", + ), + ); + assert.isFalse(childFinished.turnItems.some((item) => item.type === "error")); + assert.deepEqual( + (yield* Ref.get(state)).started.map((turn) => turn.cwd), + [sourceCwd, sourceCwd, targetCwd], + ); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry({ name }, registry, { + runEffectWorker: false, + mcpAppModelContextLayer: Layer.mock(McpAppModelContext.McpAppModelContext)({ + forThread: (threadId) => + threadId === parentId + ? Effect.succeed([]) + : Deferred.succeed(contextReadStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishContextRead)), + Effect.as([]), + ), + }), + }), + ), + ); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index ae0d5ef346d34..43283c00b646a 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -278,6 +278,7 @@ export function layerWithRegistry( SqlClient.SqlClient, MigrationError | PlatformError.PlatformError | SqlError >; + readonly mcpAppModelContextLayer?: Layer.Layer; readonly runEffectWorker?: boolean; readonly runContinuationWorker?: boolean; readonly recoverOnStartup?: boolean; @@ -403,7 +404,7 @@ export function layerWithRegistry( const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( - McpAppModelContext.layerEmpty, + options.mcpAppModelContextLayer ?? McpAppModelContext.layerEmpty, PreviewAgentActivity.layer, layerCheckpointServiceProvided, layerEventSinkProvided, diff --git a/docs/internals/overview.md b/docs/internals/overview.md index f24e3625d92f8..48e02d161780c 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -58,6 +58,10 @@ therefore means the intent committed, not that the provider, checkpoint, or othe finished. Keep external I/O out of command decisions and the database transaction. Effects tied to a lost provider process cannot simply replay; recovery retires them before admitting new work. +A worktree handoff commits its workspace binding, continuation and detach effect together. +The detach can execute immediately after commit; queuing the continuation afterward could let a +delegated child publish its result before its follow-up work exists. + The event log only keeps what live readers need: about a day of history for reconnect replay and command-id retries. [ProjectionMaintenance](../../apps/server/src/orchestration-v2/ProjectionMaintenance.ts) deletes diff --git a/docs/user/composer.md b/docs/user/composer.md index df880e49c6ca5..d816807838ad5 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -73,6 +73,9 @@ If your **Send shortcut** requires `Cmd+Enter` or `Ctrl+Enter`, that shortcut us your default follow-up behavior; add **Shift** to use the opposite action. This applies only to multiline messages when you choose the multiline send option. +When an agent moves its thread into a worktree with a continuation message, work resumes +automatically in that worktree. + Use `Cmd+Shift+Enter` on macOS or `Ctrl+Shift+Enter` on Windows and Linux to send the oldest queued message as a steer. This leaves the current draft intact and requires an active turn that supports steering. If that shortcut sends the opposite diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index b1fd02b3d3a07..eb8f7c16116f3 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3092,6 +3092,10 @@ export const OrchestrationV2Command = Schema.Union([ worktreePath: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), expectedWorktreePath: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), expectedBranch: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), + /** Record the next message atomically with a worktree handoff, before session detach. */ + worktreeContinuation: Schema.optional( + Schema.Struct({ messageId: MessageId, text: TrimmedNonEmptyString }), + ), /** Reject unless no message or run has landed on this thread. */ expectedEmpty: Schema.optional(Schema.Boolean), limitRecovery: Schema.optional(Schema.NullOr(OrchestrationV2LimitRecoveryUpdate)), diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index e925f10c8f154..e728037733366 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -429,6 +429,15 @@ const ProviderAdapterFailureReason = Schema.Literals([ "attachment_unreadable", ]); export type ProviderAdapterFailureReason = typeof ProviderAdapterFailureReason.Type; +export class ProviderAdapterSessionDetached extends Schema.TaggedError()( + "ProviderAdapterSessionDetached", + { + driver: ProviderDriverKind, + providerSessionId: ProviderSessionId, + threadId: ThreadId, + reason: Schema.Literal("workspace_changed"), + }, +) {} export class ProviderAdapterProtocolError extends Schema.TaggedError()( "ProviderAdapterProtocolError", @@ -471,6 +480,7 @@ export const ProviderAdapterV2Error = Schema.Union([ ProviderAdapterInterruptTimeoutError, ProviderAdapterRuntimeRequestResponseError, ProviderAdapterEventStreamError, + ProviderAdapterSessionDetached, ProviderAdapterProtocolError, ProviderAdapterSessionStoppedError, ]); @@ -635,8 +645,10 @@ export interface ProviderAdapterV2SessionRuntime { /** False keeps conversations alive when the provider cannot restore them after a restart. */ readonly canResumeAfterRestart?: boolean; readonly events: Stream.Stream; - /** Manager runtimes broadcast independent subscriptions; adapters may expose only a single-consumer stream. */ - readonly subscribeEvents?: Effect.Effect; + /** Manager subscriptions broadcast independently; a thread ID limits detach to its stream. */ + readonly subscribeEvents?: ( + threadId?: ThreadId, + ) => Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; /** Root-subscription probe; replacement checks opt in to subagents owned by the live process. */ readonly hasPendingBackgroundWorkForThread?: (