diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index af8fa5a95af7b..779c06b7f9176 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -189,6 +189,11 @@ export interface EventSinkV2Shape { readonly latestSequence: (input?: { readonly threadId?: ThreadId; }) => Effect.Effect; + readonly read: (input: { + readonly threadId: ThreadId; + readonly throughSequence: number; + readonly eventType: OrchestrationV2DomainEvent["type"]; + }) => Stream.Stream; readonly readByCommandId: (input: { readonly commandId: CommandId; }) => Stream.Stream; @@ -651,12 +656,10 @@ const layerBase: Layer.Layer< commandId: input.commandId, events: normalized, }); - // A Stop that changes nothing can still queue the stop of its delegated tasks. + // An eventless Stop still needs a current cutoff for its child traversal. const sequence = storedEvents.at(-1)?.sequence ?? - (input.effects.length === 0 - ? undefined - : yield* eventStore.latestSequence({ threadId: input.threadId })); + (input.effects.length === 0 ? undefined : yield* eventStore.latestSequence()); if (sequence === undefined) { return yield* Effect.die( new Error(`Command ${input.commandId} produced no orchestration events.`), @@ -977,6 +980,14 @@ const layerBase: Layer.Layer< }), ), ), + read: (input) => + eventStore + .read(input) + .pipe( + Stream.mapError( + (cause) => new EventSinkStreamError({ threadId: input.threadId, cause }), + ), + ), readByCommandId: (input) => eventStore.readByCommandId(input).pipe( Stream.mapError( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0c6ca268f1e04..2f6fc3d99a8eb 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -411,6 +411,12 @@ export interface OrchestratorV2Shape { readonly getThreadEventSequence: ( threadId: ThreadId, ) => Effect.Effect; + /** Committed history at a fixed sequence, without subscribing to live events. */ + readonly readStoredEvents: (input: { + readonly threadId: ThreadId; + readonly throughSequence: number; + readonly eventType: OrchestrationV2DomainEvent["type"]; + }) => Stream.Stream; readonly streamStoredEvents: Stream.Stream; readonly streamStoredEventsFrom: (input?: { readonly threadId?: ThreadId; @@ -10026,7 +10032,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? undefined : projection.providerThreads.find((candidate) => candidate.id === run?.providerThreadId); const hasBackgroundWork = - run?.id === projection.runs.findLast((candidate) => candidate.status !== "queued")?.id && + run?.id === latestExecutedRun(projection.runs)?.id && derivePendingBackgroundWork({ latestRun: run, providerThreads: projection.providerThreads, @@ -10562,7 +10568,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); const dispatchThreadStop = Effect.fn("Orchestrator.dispatchThreadStop")(function* ( - command: Extract, + command: Extract, events: Ref.Ref>, effects: Ref.Ref>, ) { @@ -10585,7 +10591,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) ? Effect.void : Ref.update(effects, (existing) => [...existing, delegatedTasksStopEffect(command)]); - const latestRun = projection.runs.findLast((run) => run.status !== "queued"); + const executed = latestExecutedRun(projection.runs); + const latestRun = executed ?? undefined; const target = projection.runs.findLast( (run) => @@ -10600,6 +10607,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }).length > 0 ? latestRun : undefined); + const reachedRun = target ?? executed; + if (command.throughSequence !== undefined && reachedRun !== null) { + // A failed child Stop cannot reach work started after the original parent Stop. + const hadStarted = yield* eventSink + .read({ + threadId: command.threadId, + throughSequence: command.throughSequence, + eventType: "run.created", + }) + .pipe( + Stream.concat( + eventSink.read({ + threadId: command.threadId, + throughSequence: command.throughSequence, + eventType: "run.updated", + }), + ), + Stream.runFold( + () => false, + (found, stored) => + found || + ((stored.event.type === "run.created" || stored.event.type === "run.updated") && + stored.event.runId === reachedRun.id && + stored.event.payload.status !== "queued"), + ), + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), + ), + ); + if (!hadStarted) return undefined; + } if (target !== undefined) { const interruptEvents = yield* Ref.make>([]); const interruptEffects = yield* Ref.make>([]); @@ -10636,7 +10674,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const now = yield* DateTime.now; // Record that Stop reached the turn it could not interrupt and the thread's last executed // run, so a late tool call from their agent starts nothing. - const executed = latestExecutedRun(projection.runs); const reached = new Map(); if (target !== undefined) reached.set(target.id, target); if (executed !== null && !hasLiveRun({ runs: [executed] })) reached.set(executed.id, executed); @@ -12732,6 +12769,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .latestSequence({ threadId }) .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))); }), + readStoredEvents: (input) => + eventSink + .read(input) + .pipe(Stream.mapError((cause) => new OrchestratorDomainEventStreamError({ cause }))), streamStoredEvents: eventSink.stream().pipe( Stream.mapError( (cause) => diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 9a80f8d284172..249b3781da36a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -44,6 +44,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -1162,38 +1163,83 @@ const make = Effect.gen(function* () { return { matched, unmatched: input.taskIds.filter((taskId) => !matchedIds.has(taskId)) }; }); + const stopDelegatedTasksThrough = Effect.fn("ThreadManagementService.stopDelegatedTasksThrough")( + function* ( + input: Parameters[0], + throughSequence: number | undefined, + ): Effect.fn.Return< + void, + Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError + > { + // A retry must not stop tasks delegated after the original Stop, including + // when an earlier walk stopped some children and failed on another. + // Follow-up tasks reuse a child thread, so the set walks its subtree once. + const childThreadIds = new Set( + throughSequence === undefined + ? (yield* orchestrator.getThreadRecords(input.threadId, ["subagents"])).subagents.flatMap( + (task) => + task.origin === "app_owned" && task.childThreadId !== null + ? [task.childThreadId] + : [], + ) + : yield* orchestrator + .readStoredEvents({ + threadId: input.threadId, + throughSequence, + eventType: "subagent.updated", + }) + .pipe( + Stream.filterMap((stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.origin === "app_owned" && + stored.event.payload.childThreadId !== null + ? Result.succeed(stored.event.payload.childThreadId) + : Result.failVoid, + ), + Stream.runCollect, + ), + ); + const failures: Array< + Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError + > = []; + for (const threadId of childThreadIds) { + yield* dispatch({ + type: "thread.stop", + commandId: CommandId.make(`${input.commandId}:stop:${threadId}`), + threadId, + stopDelegatedTasks: false, + ...(throughSequence === undefined ? {} : { throughSequence }), + ...(input.reason === undefined ? {} : { reason: input.reason }), + }).pipe( + Effect.andThen(stopDelegatedTasksThrough({ ...input, threadId }, throughSequence)), + Effect.catch((error) => + Effect.logWarning("Unable to stop a delegated task", { + parentThreadId: input.threadId, + threadId, + error, + }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))), + ), + ); + } + if (failures[0] !== undefined) return yield* failures[0]; + }, + ); + const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = Effect.fn( "ThreadManagementService.stopDelegatedTasks", )(function* (input) { - const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]); - const failures: Array< - Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError - > = []; - const stoppedChildThreadIds = new Set(); - for (const task of subagents) { - if (task.origin !== "app_owned" || task.childThreadId === null) continue; - const threadId = task.childThreadId; - // Follow-up tasks reuse a child thread; walk its subtree only once. - if (stoppedChildThreadIds.has(threadId)) continue; - stoppedChildThreadIds.add(threadId); - yield* dispatch({ - type: "thread.stop", - commandId: CommandId.make(`${input.commandId}:stop:${threadId}`), - threadId, - stopDelegatedTasks: false, - ...(input.reason === undefined ? {} : { reason: input.reason }), - }).pipe( - Effect.andThen(stopDelegatedTasks({ ...input, threadId })), - Effect.catch((error) => - Effect.logWarning("Unable to stop a delegated task", { - parentThreadId: input.threadId, - threadId, - error, - }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))), + const receipt = yield* receipts + .getByCommandId(input.commandId) + .pipe( + Effect.mapError( + (cause) => + new Orchestrator.OrchestratorProjectionError({ threadId: input.threadId, cause }), ), ); - } - if (failures[0] !== undefined) return yield* failures[0]; + return yield* stopDelegatedTasksThrough( + input, + Option.isSome(receipt) ? receipt.value.resultSequence : undefined, + ); }); return ThreadManagementService.of({ diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 23a52f39d33d3..a8f5ed22784e7 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -31,6 +31,7 @@ import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; import { CLAUDE_PROVIDER } from "./Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import { DispatchModeLimit } from "./DispatchModeLimit.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -611,11 +612,12 @@ it.effect("a repeated Stop that changes nothing still stops the delegated tasks" Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; + const threads = yield* ThreadManagementService.ThreadManagementService; const sql = yield* SqlClient.SqlClient; const parentThreadId = ThreadId.make("thread:stop-again"); yield* createWatchingThread(parentThreadId, 1); yield* send(parentThreadId, "first", "start_immediately"); - yield* delegate(parentThreadId, "stop-again child"); + const child = yield* delegate(parentThreadId, "stop-again child"); const now = yield* DateTime.now; const run = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; yield* projections.apply({ @@ -632,6 +634,8 @@ it.effect("a repeated Stop that changes nothing still stops the delegated tasks" threadId: parentThreadId, }); + const grandchild = yield* delegate(child, "task before fresh Stop"); + // The first Stop already disposed the delivery and ended the watch, so this one writes no // events, like Stop on a thread an older server stopped without stopping its children. const again = yield* orchestrator.dispatch({ @@ -640,6 +644,11 @@ it.effect("a repeated Stop that changes nothing still stops the delegated tasks" threadId: parentThreadId, }); assert.lengthOf(again.storedEvents, 0); + yield* threads.stopDelegatedTasks({ + threadId: parentThreadId, + commandId: CommandId.make("stop-again:second"), + }); + assert.deepEqual((yield* threadState(grandchild)).runs, ["interrupted"]); const walks = yield* sql<{ readonly command_id: string }>` SELECT command_id FROM orchestration_v2_effect_outbox WHERE effect_type = 'delegated-tasks.stop' @@ -1552,3 +1561,178 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) ); }).pipe(Effect.provide(layerTest)), ); + +it.effect("a partial delegated Stop retries only the original children that failed", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threads = yield* ThreadManagementService.ThreadManagementService; + const parentThreadId = ThreadId.make("thread:stop-partial"); + yield* createWatchingThread(parentThreadId, 6); + yield* send(parentThreadId, "work", "start_immediately"); + const rootRefused = yield* threads + .dispatch({ + type: "thread.stop", + commandId: CommandId.make("stop-root-above-limit"), + threadId: parentThreadId, + }) + .pipe( + Effect.provideService(DispatchModeLimit, { + runtimeMode: "approval-required", + interactionMode: "default", + }), + Effect.exit, + ); + assert.isTrue(Exit.isFailure(rootRefused)); + assert.deepEqual((yield* threadState(parentThreadId)).runs, ["starting"]); + const blockedChild = yield* delegate(parentThreadId, "default task"); + const advancedChild = yield* delegate(parentThreadId, "default task that advances"); + const stoppedChild = yield* delegate(parentThreadId, "plan task"); + const blockedGrandchild = yield* delegate(stoppedChild, "broader grandchild"); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("approval-child"), + threadId: stoppedChild, + runtimeMode: "approval-required", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("plan-child"), + threadId: stoppedChild, + interactionMode: "plan", + }); + const commandId = CommandId.make("stop-partial"); + yield* orchestrator.dispatch({ type: "thread.stop", commandId, threadId: parentThreadId }); + const walked = yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId }).pipe( + Effect.provideService(DispatchModeLimit, { + runtimeMode: "approval-required", + interactionMode: "plan", + }), + Effect.exit, + ); + assert.isTrue(Exit.isFailure(walked)); + assert.deepEqual((yield* threadState(stoppedChild)).runs, ["interrupted"]); + assert.deepEqual((yield* threadState(blockedChild)).runs, ["starting"]); + assert.deepEqual((yield* threadState(blockedGrandchild)).runs, ["starting"]); + + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("independent-child-stop"), + threadId: advancedChild, + }); + yield* send(blockedChild, "queued after failed Stop", "queue_after_active"); + yield* orchestrator.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make("cancel-after-failed-stop"), + threadId: blockedChild, + runId: (yield* orchestrator.getThreadProjection(blockedChild)).runs.at(-1)!.id, + }); + yield* send(advancedChild, "new work before failed Stop retry", "start_immediately"); + yield* send(stoppedChild, "new child work", "start_immediately"); + const newGrandchild = yield* delegate(stoppedChild, "new grandchild"); + yield* send(parentThreadId, "new parent work", "start_immediately"); + const newChild = yield* delegate(parentThreadId, "new child"); + yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId }); + assert.deepEqual((yield* threadState(blockedChild)).runs, ["interrupted", "cancelled"]); + assert.deepEqual((yield* threadState(blockedGrandchild)).runs, ["interrupted"]); + assert.deepEqual((yield* threadState(stoppedChild)).runs, ["interrupted", "starting"]); + assert.deepEqual((yield* threadState(advancedChild)).runs, ["interrupted", "starting"]); + assert.deepEqual((yield* threadState(newChild)).runs, ["starting"]); + assert.deepEqual((yield* threadState(newGrandchild)).runs, ["starting"]); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("a bounded Stop ignores cancelled queues and preserves newer completed work", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const completeStartedRun = (threadId: ThreadId) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const run = (yield* orchestrator.getThreadProjection(threadId)).runs.findLast( + (candidate) => candidate.status === "starting", + )!; + yield* projections.apply({ + id: EventId.make(`event:${run.id}:completed`), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", startedAt: now, completedAt: now }, + }); + }); + const retryStop = (threadId: ThreadId, throughSequence: number) => + orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make(`retry-stop:${threadId}`), + threadId, + throughSequence, + }); + + const cancelledQueue = ThreadId.make("thread:stop-retry-cancelled-queue"); + yield* createWatchingThread(cancelledQueue, 13); + yield* send(cancelledQueue, "work", "start_immediately"); + const cancelledQueueStop = yield* orchestrator.getThreadEventSequence(cancelledQueue); + yield* send(cancelledQueue, "queued after Stop", "queue_after_active"); + yield* orchestrator.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make("cancel-queued-after-stop"), + threadId: cancelledQueue, + runId: (yield* orchestrator.getThreadProjection(cancelledQueue)).runs.at(-1)!.id, + }); + yield* completeStartedRun(cancelledQueue); + const providerThread = (yield* orchestrator.getThreadProjection(cancelledQueue)) + .providerThreads[0]!; + yield* projections.apply({ + id: EventId.make("event:stop-retry:original-background"), + type: "provider-thread.updated", + threadId: cancelledQueue, + occurredAt: yield* DateTime.now, + payload: { + ...providerThread, + pendingBackgroundTasks: [{ taskId: "original-background", kind: "command" }], + }, + }); + const originalRun = (yield* orchestrator.getThreadProjection(cancelledQueue)).runs[0]!; + const completedAt = yield* DateTime.now; + yield* projections.apply({ + id: EventId.make("event:stop-retry:completed-provider-turn"), + type: "provider-turn.updated", + threadId: cancelledQueue, + occurredAt: completedAt, + payload: { + id: ProviderTurnId.make("turn:stop-retry:completed"), + providerThreadId: providerThread.id, + nodeId: originalRun.rootNodeId!, + runAttemptId: originalRun.activeAttemptId!, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: completedAt, + completedAt, + }, + }); + yield* retryStop(cancelledQueue, cancelledQueueStop); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(cancelledQueue)).providerThreads[0]! + .pendingBackgroundTasks, + [], + ); + assert.deepEqual(yield* threadState(cancelledQueue), { + runs: ["completed", "cancelled"], + watched: [], + }); + + const newerRun = ThreadId.make("thread:stop-retry-newer-run"); + yield* createWatchingThread(newerRun, 14); + yield* send(newerRun, "work", "start_immediately"); + const newerRunStop = yield* orchestrator.getThreadEventSequence(newerRun); + yield* completeStartedRun(newerRun); + yield* send(newerRun, "work after Stop", "start_immediately"); + yield* completeStartedRun(newerRun); + yield* retryStop(newerRun, newerRunStop); + assert.deepEqual(yield* threadState(newerRun), { + runs: ["completed", "completed"], + watched: [14], + }); + }).pipe(Effect.provide(layerTest)), +); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index f4e8303972267..3ea7a4dbf0a0b 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3292,6 +3292,18 @@ const OrchestrationV2InternalCommand = Schema.Union([ }), ), }), + /** + * A delegated Stop bounded by the parent Stop's receipt. Clients send the + * public `thread.stop`, whose decoding drops this bound. + */ + Schema.Struct({ + type: Schema.Literal("thread.stop"), + commandId: CommandId, + threadId: ThreadId, + reason: Schema.optional(Schema.String), + stopDelegatedTasks: Schema.optional(Schema.Boolean), + throughSequence: Schema.optional(NonNegativeInt), + }), /** * Internal so no client can mark a request saved without the value being * stored. diff --git a/packages/contracts/src/rpc.test.ts b/packages/contracts/src/rpc.test.ts index 718cd3e50f84d..5e7ba74eabfb2 100644 --- a/packages/contracts/src/rpc.test.ts +++ b/packages/contracts/src/rpc.test.ts @@ -35,6 +35,43 @@ describe("WebSocket RPC contracts", () => { ).toBe(true); }); + it("keeps server-internal Stop retry bounds off the dispatch surface", () => { + const dispatchCommand = WsRpcGroup.requests.get(ORCHESTRATION_V2_WS_METHODS.dispatchCommand); + if (dispatchCommand === undefined) throw new Error("dispatchCommand is not registered"); + const decode = Schema.decodeUnknownSync(dispatchCommand.payloadSchema); + + expect( + decode({ + type: "thread.stop", + commandId: "stop-1", + threadId: "thread-1", + reason: "User stopped the thread.", + throughSequence: 0, + }), + ).toStrictEqual({ + type: "thread.stop", + commandId: "stop-1", + threadId: "thread-1", + reason: "User stopped the thread.", + }); + expect( + decode({ + type: "run.interrupt", + commandId: "stop-1", + threadId: "thread-1", + runId: "run-1", + reason: "User stopped the thread.", + throughSequence: 0, + }), + ).toStrictEqual({ + type: "run.interrupt", + commandId: "stop-1", + threadId: "thread-1", + runId: "run-1", + reason: "User stopped the thread.", + }); + }); + it("lets the server pick a new worktree's folder", () => { const createWorktree = WsRpcGroup.requests.get(WS_METHODS.vcsCreateWorktree); if (createWorktree === undefined) throw new Error("vcs.createWorktree is not registered");