From aa6e3a0b65764172556f3ee02fa2c01ff013b023 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 16:58:43 +0000 Subject: [PATCH 1/3] test(server): reproduce #30989 Force stop leaving a stranded run running A run still projected running whose provider turn already completed, on a live session shared with another working thread, makes Force stop dispatch run.interrupt instead of thread.background-work.settle, so the run is never settled. Co-Authored-By: Claude Opus 5.5 --- .../ThreadManagementService.test.ts | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 9285f72d46f56..f2fa3c2d01009 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -628,6 +628,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, @@ -692,7 +703,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), From 1dbbc23a5d0d153a1e027fc53337a80e9db087f3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 17:07:35 +0000 Subject: [PATCH 2/3] fix(server): Force stop settles a running run whose provider turn already completed Force stop only settled an active run through a running provider turn. A run still projected running after its turn completed fell through to run.interrupt, which on a live (e.g. shared) session just queued another provider interrupt that could time out and leave the run running. Force stop now settles any turn of a running run's active attempt, the same turn run.interrupt itself would pick. Runs still preparing or starting (pending provider-turn.start or restart) keep the run.interrupt path, which cancels those effects. Fixes pingdotgg/t3code#30989 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/ThreadManagementService.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index ed2935fedbf12..9965fb6771d40 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1159,7 +1159,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}`, From 4b895063337f54c7022c6afb040e5e8dc0f2669f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 17:28:03 +0000 Subject: [PATCH 3/3] fix(server): Force stop records the Stop on a stranded run it settles Settling a running run that no Stop reached, which only Force stop does, skipped run.interrupt's bookkeeping: no interrupt-request marker, so a late agent PR watch was accepted, and its delegated-completion cohort stayed open, so a child finishing later woke the thread into a new run. thread.background-work.settle now writes the marker and stops the cohorts when it settles a running run without one, sharing the cohort disposal with run.interrupt. Runs a Stop already reached, including held-queue Stops, are untouched. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 47 ++++++++++---- .../src/orchestration-v2/ThreadStop.test.ts | 65 +++++++++++++++++++ 2 files changed, 99 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b4bdbd705c290..2dc49b0f1a355 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10782,7 +10782,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; @@ -10831,6 +10831,32 @@ 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 completionCohortRunId = + projection.messages.find((candidate) => candidate.id === run.userMessageId) + ?.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, + }); + } + }); + const holdStoppedThread = Effect.fn("Orchestrator.holdStoppedThread")(function* (input: { readonly command: Extract< OrchestrationV2ServerCommand, @@ -11133,6 +11159,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], @@ -11371,18 +11403,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, - }); - } - }); + : stopDelegatedCompletionCohorts({ command, events, run, now }); const emitEvent = emit(events, command); const interruptRequestItem: OrchestrationV2TurnItem = { 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)), +);