diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3157f55cacfa6..9838638f2206b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11251,7 +11251,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const markStoppedRun = Effect.fnUntraced(function* (input: { readonly command: Extract< OrchestrationV2ServerCommand, - { readonly type: "run.interrupt" | "thread.stop" } + { readonly type: "run.interrupt" | "thread.stop" | "thread.background-work.settle" } >; readonly events: Ref.Ref>; readonly thread: OrchestrationV2AppThread; @@ -11300,6 +11300,36 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + const stopDelegatedCompletionCohorts = Effect.fnUntraced(function* (input: { + readonly command: Extract< + OrchestrationV2ServerCommand, + { readonly type: "run.interrupt" | "thread.background-work.settle" } + >; + readonly events: Ref.Ref>; + readonly run: OrchestrationV2Run; + readonly now: DateTime.Utc; + }) { + const { command, events, run, now } = input; + const projection = yield* getProjectionWithPendingEvents(command.threadId, events); + const delegatedCompletion = projection.messages.find( + (candidate) => candidate.id === run.userMessageId, + )?.delegatedCompletion; + const completionCohortRunId = delegatedCompletion?.parentRunId ?? run.id; + for (const parentRunId of new Set([completionCohortRunId, run.id])) { + yield* disposeDelegatedCompletionCohort({ + command, + events, + projection: yield* getProjectionWithPendingEvents(command.threadId, events), + parentRunId, + disposition: "stopped", + now, + ...(parentRunId !== run.id && delegatedCompletion !== undefined + ? { interruptedWakeTaskIds: delegatedCompletion.taskIds } + : {}), + }); + } + }); + const holdStoppedThread = Effect.fn("Orchestrator.holdStoppedThread")(function* (input: { readonly command: Extract< OrchestrationV2ServerCommand, @@ -11602,6 +11632,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return; const now = yield* DateTime.now; if (stopped.providerTurn !== undefined && stoppedRun.status === "running") { + // Only Force stop settles a running run no Stop reached; it records that Stop itself. + if (!(yield* stopReachedRun(command, command.threadId, stoppedRun.id))) { + const { thread } = yield* getProjectionWithPendingEvents(command.threadId, events); + yield* markStoppedRun({ command, events, thread, run: stoppedRun, now }); + yield* stopDelegatedCompletionCohorts({ command, events, run: stoppedRun, now }); + } const output = yield* projectionStore .getThreadRecords(command.threadId, ["messages"], { messageRunIds: [stoppedRun.id], @@ -11840,21 +11876,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cohortRunIds: [completionCohortRunId], now, }) - : Effect.gen(function* () { - for (const parentRunId of new Set([completionCohortRunId, run.id])) { - yield* disposeDelegatedCompletionCohort({ - command, - events, - projection: yield* getProjectionWithPendingEvents(command.threadId, events), - parentRunId, - disposition: "stopped", - now, - ...(parentRunId !== run.id && completionMessage?.delegatedCompletion !== undefined - ? { interruptedWakeTaskIds: completionMessage.delegatedCompletion.taskIds } - : {}), - }); - } - }); + : stopDelegatedCompletionCohorts({ command, events, run, now }); const emitEvent = emit(events, command); const interruptRequestItem: OrchestrationV2TurnItem = { diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 742616f27bd3e..5bfaaf3ad1489 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -732,6 +732,17 @@ it.effect.each([ }, teardown: ["detach"], }, + { + name: "settles a running run whose provider turn already completed in a shared process", + others: ["thread:thread-management:force-stop:other"], + killed: false, + strandedRun: true, + expected: { + outcome: "shared", + sharedWithThreadIds: ["thread:thread-management:force-stop:other"], + }, + teardown: ["detach"], + }, { name: "detaches when a sibling starts a turn just before the session is released", others: [] as ReadonlyArray, @@ -796,7 +807,7 @@ it.effect.each([ id: "provider-turn:force-stop", providerThreadId, runAttemptId: "attempt", - status: "settledRun" in row ? "completed" : "running", + status: "settledRun" in row || "strandedRun" in row ? "completed" : "running", }, ], } as unknown as OrchestrationV2ThreadProjection), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index de12768ac7b17..cf46883d41e33 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1234,7 +1234,8 @@ const make = Effect.gen(function* () { const active = isActiveRun(run); const providerTurn = records.providerTurns.findLast( (turn) => - turn.runAttemptId === run.activeAttemptId && (!active || turn.status === "running"), + turn.runAttemptId === run.activeAttemptId && + (!active || run.status === "running" || turn.status === "running"), ); const commandId = CommandId.make( `server:force-stop:${run.id}:${yield* Clock.currentTimeMillis}`, diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 6b3ea08a86449..6c5a9f80384db 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1742,3 +1742,68 @@ it.effect.each(["Stop", "Force stop"] as const)( ); }).pipe(Effect.provide(layerTest)), ); + +it.effect("Force stop alone stops a stranded run's delegated cohort and later agent watches", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make("thread:force-stop-stranded"); + yield* createWatchingThread(threadId, 8); + yield* send(threadId, "parent", "start_immediately"); + yield* delegate(threadId, "stranded child"); + const now = yield* DateTime.now; + const started = yield* orchestrator.getThreadProjection(threadId); + const run = started.runs[0]!; + const attempt = started.attempts[0]!; + const apply = (event: Omit) => + projections.apply({ + ...event, + id: EventId.make(`event:force-stop-stranded:${event.type}`), + threadId, + } as OrchestrationV2DomainEvent); + yield* apply({ + type: "run.updated", + runId: run.id, + occurredAt: now, + payload: { + ...run, + status: "running", + startedAt: now, + delegatedCompletion: { disposition: "open", nextGeneration: 1, delivery: null }, + }, + }); + yield* apply({ + type: "run-attempt.updated", + occurredAt: now, + payload: { ...attempt, status: "completed", startedAt: now, completedAt: now }, + }); + yield* apply({ + type: "provider-turn.updated", + occurredAt: now, + payload: { + id: ProviderTurnId.make("provider-turn:force-stop-stranded"), + providerThreadId: run.providerThreadId!, + nodeId: run.rootNodeId!, + runAttemptId: attempt.id, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: now, + completedAt: now, + }, + }); + + const threads = yield* ThreadManagementService.ThreadManagementService; + yield* threads.forceStopThread({ threadId, runId: run.id }); + const stopped = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + stopped.runs.map((candidate) => [ + candidate.status, + candidate.delegatedCompletion?.disposition, + ]), + [["interrupted", "stopped"]], + ); + assert.equal(stopped.subagents[0]?.completionDelivery?.state, "disposed"); + assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 9)))); + }).pipe(Effect.provide(layerTest)), +);