fix(server): Force stop settles a running run whose provider turn already completed

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<Array<OrchestrationV2DomainEvent>>;
     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<Array<OrchestrationV2DomainEvent>>;
+    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 = {

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<string>,
@@ -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),

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}`,

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<OrchestrationV2DomainEvent, "id" | "threadId">) =>
+      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)),
+);