fix(server): Stop reaches provider turns accepted during startup

apps/server/src/orchestration-v2/EffectOutbox.ts

@@ -44,6 +44,11 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
     type: Schema.Literal("provider-turn.start"),
     runId: RunId,
   }),
+  Schema.Struct({
+    type: Schema.Literal("provider-turn.interrupt-start"),
+    runId: RunId,
+    attemptId: RunAttemptId,
+  }),
   Schema.Struct({
     type: Schema.Literal("provider-turn.interrupt"),
     providerSessionId: ProviderSessionId,
@@ -138,6 +143,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [
 export const PROCESS_BOUND_EFFECT_TYPES = [
   "provider-turn.start",
   "provider-turn.interrupt",
+  "provider-turn.interrupt-start",
   "provider-turn.steer",
   "provider-turn.restart",
   "runtime-request.respond",
@@ -209,6 +215,7 @@ export interface EffectOutboxV2Shape {
   ) => Effect.Effect<ReadonlyArray<OrchestrationEffectV2>, EffectOutboxError>;
   readonly cancelUnsettled: (input: {
     readonly threadId: ThreadId;
+    readonly runId?: RunId;
     readonly effectTypes: ReadonlyArray<OrchestrationEffectRequestV2["type"]>;
     readonly reason: string;
   }) => Effect.Effect<ReadonlyArray<string>, EffectOutboxError>;
@@ -437,7 +444,7 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
           ),
         ),
       cancelUnsettled: Effect.fn("EffectOutbox.cancelUnsettled")(
-        function* ({ threadId, effectTypes, reason }) {
+        function* ({ threadId, runId, effectTypes, reason }) {
           if (effectTypes.length === 0) return [];
           const now = DateTime.formatIso(yield* DateTime.now);
           const rows = yield* sql<{ readonly effect_id: string }>`
@@ -451,6 +458,7 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
             WHERE thread_id = ${threadId}
               AND status IN ('pending', 'running')
               AND effect_type IN ${sql.in(effectTypes)}
+              AND (${runId ?? null} IS NULL OR json_extract(payload_json, '$.runId') = ${runId ?? null})
             RETURNING effect_id
           `;
           return rows.map(({ effect_id }) => effect_id);

apps/server/src/orchestration-v2/EffectWorker.test.ts

@@ -31,6 +31,7 @@ import * as RunFinalizationService from "./RunFinalizationService.ts";
 import * as ProviderSessionManager from "./ProviderSessionManager.ts";
 import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
 import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
+import * as RunExecutionService from "./RunExecutionService.ts";
 import * as RuntimeRequestService from "./RuntimeRequestService.ts";
 import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts";
 import * as ThreadManagementService from "./ThreadManagementService.ts";
@@ -88,6 +89,7 @@ function layerExecutorFor(input: {
 }) {
   const record = (event: string) => Ref.update(input.events, (events) => [...events, event]);
   const layerDependencies = Layer.mergeAll(
+    Layer.mock(RunExecutionService.RunExecutionServiceV2)({}),
     Layer.succeed(
       ProviderTurnControlService.ProviderTurnControlServiceV2,
       ProviderTurnControlService.ProviderTurnControlServiceV2.of({

apps/server/src/orchestration-v2/EffectWorker.ts

@@ -23,6 +23,7 @@ import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
 import * as ProviderSessionManager from "./ProviderSessionManager.ts";
 import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
 import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
+import * as RunExecutionService from "./RunExecutionService.ts";
 import * as RuntimeRequestService from "./RuntimeRequestService.ts";
 import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts";
 import * as ThreadManagementService from "./ThreadManagementService.ts";
@@ -76,6 +77,7 @@ export const layerExecutor: Layer.Layer<
   | CheckpointRollbackService.CheckpointRollbackServiceV2
   | ProviderTurnControlService.ProviderTurnControlServiceV2
   | ProviderTurnStartService.ProviderTurnStartServiceV2
+  | RunExecutionService.RunExecutionServiceV2
   | RuntimeRequestService.RuntimeRequestServiceV2
   | ThreadTitleRegenerationService.ThreadTitleRegenerationService
   | ThreadManagementService.ThreadManagementService
@@ -89,6 +91,7 @@ export const layerExecutor: Layer.Layer<
     const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
     const providerTurnControl = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
     const providerTurnStart = yield* ProviderTurnStartService.ProviderTurnStartServiceV2;
+    const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
     const runtimeRequests = yield* RuntimeRequestService.RuntimeRequestServiceV2;
     const threadTitleRegeneration =
       yield* ThreadTitleRegenerationService.ThreadTitleRegenerationService;
@@ -151,6 +154,49 @@ export const layerExecutor: Layer.Layer<
                     }),
                 ),
               );
+          case "provider-turn.interrupt-start": {
+            const request = effect.request;
+            return runExecution
+              .interruptOwnedStart({
+                threadId: effect.threadId,
+                runId: request.runId,
+                attemptId: request.attemptId,
+              })
+              .pipe(
+                Effect.map((target) => ({ unresponsive: false, target }) as const),
+                Effect.catchIf(
+                  (cause) => cause.unresponsive,
+                  () => Effect.succeed({ unresponsive: true, target: undefined } as const),
+                ),
+                Effect.andThen(({ unresponsive, target }) =>
+                  unresponsive
+                    ? threads.dispatch({
+                        type: "run.interrupt.unresponsive",
+                        commandId: CommandId.make(`${effect.id}:unresponsive`),
+                        threadId: effect.threadId,
+                        runId: request.runId,
+                        attemptId: request.attemptId,
+                      })
+                    : target === undefined
+                      ? Effect.void
+                      : threads.dispatch({
+                          type: "thread.background-work.settle",
+                          commandId: CommandId.make(`${effect.id}:background-work-settled`),
+                          threadId: effect.threadId,
+                          providerThreadId: target.providerThreadId,
+                          providerTurnId: target.providerTurnId,
+                        }),
+                ),
+                Effect.mapError(
+                  (cause) =>
+                    new OrchestrationEffectExecutionError({
+                      effectId: effect.id,
+                      effectType: request.type,
+                      cause,
+                    }),
+                ),
+              );
+          }
           case "provider-turn.interrupt": {
             const request = effect.request;
             return providerTurnControl

apps/server/src/orchestration-v2/EventSink.ts

@@ -171,6 +171,7 @@ export interface EventSinkV2Shape {
     readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
     readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
     readonly cancelUnsettledEffects?: {
+      readonly runId?: RunId;
       readonly effectTypes: ReadonlyArray<EffectOutbox.OrchestrationEffectRequestV2["type"]>;
       readonly reason: string;
     };

apps/server/src/orchestration-v2/FoundationPersistence.test.ts

@@ -2251,12 +2251,31 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
         const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
         const workerFiber = yield* worker.runOnce.pipe(Effect.forkChild);
         yield* Deferred.await(started);
+        const replacementId = `${effectId}:replacement`;
+        yield* outbox.enqueue([
+          {
+            id: replacementId,
+            commandId,
+            threadId,
+            request: { type: "provider-turn.start", runId: RunId.make("run:replacement") },
+          },
+        ]);
         const cancelledEffectIds = yield* outbox.cancelUnsettled({
           threadId,
-          effectTypes: ["provider-turn.start"],
+          runId: RunId.make("run:foundation-cancel-running-effect"),
+          effectTypes: ["provider-turn.start", "provider-turn.restart"],
           reason: "The owning run was interrupted.",
         });
         assert.deepEqual(cancelledEffectIds, [effectId]);
+        assert.equal(Option.getOrThrow(yield* outbox.get(replacementId)).status, "pending");
+        yield* outbox.signalCancellations(
+          yield* outbox.cancelUnsettled({
+            threadId,
+            runId: RunId.make("run:replacement"),
+            effectTypes: ["provider-turn.start"],
+            reason: "Test cleanup.",
+          }),
+        );
         yield* outbox.signalCancellations(cancelledEffectIds);
         assert.isTrue(yield* Fiber.join(workerFiber));
         yield* Deferred.await(interrupted);