# R0 and R6: immutable source inventory

R0/R6 require source inventory; no runtime capture is claimed. R6 deterministic race harness was optional and was not run.

R0: thread.stop emits a stop only for nondeleted threads with delegation enabled and an app-owned child with nonnull child ID. run.interrupt emits a stop when holdQueue is true, including empty-child no-op stops. Plain interrupts do not. MCP directly invokes the recursive service and bypasses the worker wrapper. Native children and null child IDs are skipped; duplicate child IDs are deduplicated.

R6: cancellation Deferred is registered before the durable reread. A cancelled row clears registration and returns before executor invocation or the new timeout. This branch is byte-identical between revisions. Already cancelled pending rows are not claimable.

The stop-only timeout wraps each executor attempt, inside the cancellation race. The default maxAttempts=5, leaseDurationMs=30000 and 100*2**(attempt-1) retry delays are unchanged. A running lease is not reclaimed in-process. Thread-title work occupies a separate lane.

The 14 non-stop identity types are provider-runtime.continue, provider-session.detach, provider-turn.start, provider-turn.interrupt, provider-turn.steer, provider-turn.restart, runtime-request.respond, provider-thread.rollback, checkpoint.capture, scheduled-tasks.cleanup, terminal.cleanup, preview.cleanup, attachment.cleanup, thread-title.generate.

The bound requires cooperative interruption. Synchronous JS blockage, uninterruptible executor regions or hanging finalizers can delay settlement. Direct service calls have no new bound.

## apps/server/src/orchestration-v2/Orchestrator.ts

Whole-file identical: True.

[96fc7b9551](https://git.cafe/pingdotgg/t3code/blob/96fc7b955196637605cbf3b415152ce4dd0e2294/apps/server/src/orchestration-v2/Orchestrator.ts), SHA256 `f009a1fda34a5bb0136b004e51983e5a4cb55afd46f55a9dc8976c5a9f23725d`

```ts
550:   return {
551:     id: `effect:${commandId}:thread-title.generate`,
552:     commandId,
553:     threadId,
554:     request: { type: "thread-title.generate", kind },
555:   };
556: }
557: 
558: function delegatedTasksStopEffect(command: {
559:   readonly commandId: CommandId;
560:   readonly threadId: ThreadId;
561:   readonly reason?: string | undefined;
562: }): PendingOrchestrationEffectV2 {
563:   return {
564:     id: `effect:${command.commandId}:delegated-tasks.stop`,
565:     commandId: command.commandId,
566:     threadId: command.threadId,
567:     request: {
568:       type: "delegated-tasks.stop",
569:       ...(command.reason === undefined ? {} : { reason: command.reason }),
570:     },
571:   };
572: }
573: 
574: const WORKSPACE_PREPARATION_INPUT = "Preparing workspace";
575: const RUNTIME_REQUEST_DELIVERY_FAILED_MESSAGE =
10562:     return interrupted.value;
10563:   });
10564: 
10565:   const dispatchThreadStop = Effect.fn("Orchestrator.dispatchThreadStop")(function* (
10566:     command: Extract<OrchestrationV2ServerCommand, { readonly type: "thread.stop" }>,
10567:     events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
10568:     effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
10569:   ) {
10570:     const projection = yield* projectionStore
10571:       .getThreadRecords(command.threadId, ["runs", "providerThreads", "turnItems", "subagents"], {
10572:         turnItemTypes: ["command_execution", "dynamic_tool", "subagent"],
10573:         turnItemStatuses: ["pending", "running", "waiting"],
10574:       })
10575:       .pipe(
10576:         Effect.mapError(
10577:           (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
10578:         ),
10579:       );
10580:     if (projection.thread.deletedAt !== null) return undefined;
10581:     // Queued after the interrupt, since a thread's effects run in order.
10582:     const stopDelegatedTasks =
10583:       command.stopDelegatedTasks === false ||
10584:       !projection.subagents.some(
10585:         (task) => task.origin === "app_owned" && task.childThreadId !== null,
10586:       )
10587:         ? Effect.void
10588:         : Ref.update(effects, (existing) => [...existing, delegatedTasksStopEffect(command)]);
10589:     const latestRun = projection.runs.findLast((run) => run.status !== "queued");
10590:     const target =
10591:       projection.runs.findLast(
10592:         (run) =>
10593:           run.status === "preparing" || run.status === "starting" || run.status === "running",
10594:       ) ??
10595:       (derivePendingBackgroundWork({
10596:         latestRun,
10597:         providerThreads: projection.providerThreads,
10598:         turnItems: projection.turnItems,
10599:         activeProviderThreadId: projection.thread.activeProviderThreadId,
10600:         runs: projection.runs,
10601:       }).length > 0
10602:         ? latestRun
10603:         : undefined);
10604:     if (target !== undefined) {
10605:       const interruptEvents = yield* Ref.make<Array<OrchestrationV2DomainEvent>>([]);
10606:       const interruptEffects = yield* Ref.make<Array<PendingOrchestrationEffectV2>>([]);
10607:       const interrupted = yield* Effect.exit(
10608:         dispatchRunInterrupt(
10609:           {
10610:             type: "run.interrupt",
10611:             commandId: command.commandId,
10612:             threadId: command.threadId,
10613:             runId: target.id,
10614:             holdQueue: true,
10615:             ...(command.reason === undefined ? {} : { reason: command.reason }),
10616:           },
10617:           interruptEvents,
10618:           interruptEffects,
10619:         ),
10620:       );
10621:       // A failed interrupt may have written part of its events, so they are kept apart.
10622:       if (Exit.isSuccess(interrupted)) {
10623:         const written = yield* Ref.get(interruptEvents);
10624:         const enqueued = yield* Ref.get(interruptEffects);
10625:         yield* Ref.update(events, (existing) => [...existing, ...written]);
10626:         yield* Ref.update(effects, (existing) => [...existing, ...enqueued]);
10627:         yield* cancelIdleOrphans({ command, events, projection });
10628:         yield* stopDelegatedTasks;
10629:         return interrupted.value;
10630:       }
10631:       yield* Effect.logWarning("thread.stop could not interrupt the running turn", {
10632:         threadId: command.threadId,
10633:         runId: target.id,
10634:         cause: interrupted.cause,
10635:       });
10636:     }
10637:     const now = yield* DateTime.now;
10638:     // Record that Stop reached the turn it could not interrupt and the thread's last executed
10639:     // run, so a late tool call from their agent starts nothing.
10640:     const executed = latestExecutedRun(projection.runs);
10641:     const reached = new Map<RunId, OrchestrationV2Run>();
10642:     if (target !== undefined) reached.set(target.id, target);
10643:     if (executed !== null && !hasLiveRun({ runs: [executed] })) reached.set(executed.id, executed);
10644:     for (const run of reached.values()) {
10645:       yield* markStoppedRun({ command, events, thread: projection.thread, run, now });
10646:     }
10647:     yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now });
10648:     yield* cancelIdleOrphans({ command, events, projection });
10649:     yield* stopDelegatedTasks;
10650:     return undefined;
10651:   });
11925:         yield* dispatchRuntimeRequestRespond(command, events, effects);
11926:         break;
11927:       case "thread.user-input.dismiss":
11928:         yield* dispatchThreadUserInputDismiss(command, events, effects);
11929:         break;
11930:       case "subagent.stop":
11931:         yield* dispatchSubagentStop(command, events, effects);
11932:         break;
11933:       case "run.interrupt":
11934:         cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects);
11935:         if (command.holdQueue === true) {
11936:           // Stop must still commit when a child can't be read, so a failed repair's
11937:           // partial events are dropped.
11938:           const repairEvents = yield* Ref.make(yield* Ref.get(events));
11939:           const repaired = yield* Effect.exit(
11940:             loadProjectionForCommand(command, ["subagents", "turnItems"], {
11941:               turnItemTypes: ["subagent"],
11942:               turnItemStatuses: ["pending", "running", "waiting"],
11943:             }).pipe(
11944:               Effect.flatMap((projection) =>
11945:                 cancelIdleOrphans({ command, events: repairEvents, projection }),
11946:               ),
11947:             ),
11948:           );
11949:           if (Exit.isSuccess(repaired)) yield* Ref.set(events, yield* Ref.get(repairEvents));
11950:           else
11951:             yield* Effect.logWarning("Stop could not cancel orphaned delegated follow-ups", {
11952:               threadId: command.threadId,
11953:               cause: repaired.cause,
11954:             });
11955:           yield* Ref.update(effects, (existing) => [
11956:             ...existing,
11957:             delegatedTasksStopEffect(command),
11958:           ]);
11959:         }
11960:         break;
11961:       case "thread.stop":
11962:         cancelUnsettledEffects = yield* dispatchThreadStop(command, events, effects);
11963:         break;
11964:       case "queued-message.promote-to-steer":
11965:         cancelUnsettledEffects = yield* dispatchQueuedMessagePromoteToSteer(
```

[802efb97da](https://git.cafe/pingdotgg/t3code/blob/802efb97da70e935d2c55ff2be6caf1093ffdb32/apps/server/src/orchestration-v2/Orchestrator.ts), SHA256 `f009a1fda34a5bb0136b004e51983e5a4cb55afd46f55a9dc8976c5a9f23725d`

```ts
550:   return {
551:     id: `effect:${commandId}:thread-title.generate`,
552:     commandId,
553:     threadId,
554:     request: { type: "thread-title.generate", kind },
555:   };
556: }
557: 
558: function delegatedTasksStopEffect(command: {
559:   readonly commandId: CommandId;
560:   readonly threadId: ThreadId;
561:   readonly reason?: string | undefined;
562: }): PendingOrchestrationEffectV2 {
563:   return {
564:     id: `effect:${command.commandId}:delegated-tasks.stop`,
565:     commandId: command.commandId,
566:     threadId: command.threadId,
567:     request: {
568:       type: "delegated-tasks.stop",
569:       ...(command.reason === undefined ? {} : { reason: command.reason }),
570:     },
571:   };
572: }
573: 
574: const WORKSPACE_PREPARATION_INPUT = "Preparing workspace";
575: const RUNTIME_REQUEST_DELIVERY_FAILED_MESSAGE =
10562:     return interrupted.value;
10563:   });
10564: 
10565:   const dispatchThreadStop = Effect.fn("Orchestrator.dispatchThreadStop")(function* (
10566:     command: Extract<OrchestrationV2ServerCommand, { readonly type: "thread.stop" }>,
10567:     events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
10568:     effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
10569:   ) {
10570:     const projection = yield* projectionStore
10571:       .getThreadRecords(command.threadId, ["runs", "providerThreads", "turnItems", "subagents"], {
10572:         turnItemTypes: ["command_execution", "dynamic_tool", "subagent"],
10573:         turnItemStatuses: ["pending", "running", "waiting"],
10574:       })
10575:       .pipe(
10576:         Effect.mapError(
10577:           (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
10578:         ),
10579:       );
10580:     if (projection.thread.deletedAt !== null) return undefined;
10581:     // Queued after the interrupt, since a thread's effects run in order.
10582:     const stopDelegatedTasks =
10583:       command.stopDelegatedTasks === false ||
10584:       !projection.subagents.some(
10585:         (task) => task.origin === "app_owned" && task.childThreadId !== null,
10586:       )
10587:         ? Effect.void
10588:         : Ref.update(effects, (existing) => [...existing, delegatedTasksStopEffect(command)]);
10589:     const latestRun = projection.runs.findLast((run) => run.status !== "queued");
10590:     const target =
10591:       projection.runs.findLast(
10592:         (run) =>
10593:           run.status === "preparing" || run.status === "starting" || run.status === "running",
10594:       ) ??
10595:       (derivePendingBackgroundWork({
10596:         latestRun,
10597:         providerThreads: projection.providerThreads,
10598:         turnItems: projection.turnItems,
10599:         activeProviderThreadId: projection.thread.activeProviderThreadId,
10600:         runs: projection.runs,
10601:       }).length > 0
10602:         ? latestRun
10603:         : undefined);
10604:     if (target !== undefined) {
10605:       const interruptEvents = yield* Ref.make<Array<OrchestrationV2DomainEvent>>([]);
10606:       const interruptEffects = yield* Ref.make<Array<PendingOrchestrationEffectV2>>([]);
10607:       const interrupted = yield* Effect.exit(
10608:         dispatchRunInterrupt(
10609:           {
10610:             type: "run.interrupt",
10611:             commandId: command.commandId,
10612:             threadId: command.threadId,
10613:             runId: target.id,
10614:             holdQueue: true,
10615:             ...(command.reason === undefined ? {} : { reason: command.reason }),
10616:           },
10617:           interruptEvents,
10618:           interruptEffects,
10619:         ),
10620:       );
10621:       // A failed interrupt may have written part of its events, so they are kept apart.
10622:       if (Exit.isSuccess(interrupted)) {
10623:         const written = yield* Ref.get(interruptEvents);
10624:         const enqueued = yield* Ref.get(interruptEffects);
10625:         yield* Ref.update(events, (existing) => [...existing, ...written]);
10626:         yield* Ref.update(effects, (existing) => [...existing, ...enqueued]);
10627:         yield* cancelIdleOrphans({ command, events, projection });
10628:         yield* stopDelegatedTasks;
10629:         return interrupted.value;
10630:       }
10631:       yield* Effect.logWarning("thread.stop could not interrupt the running turn", {
10632:         threadId: command.threadId,
10633:         runId: target.id,
10634:         cause: interrupted.cause,
10635:       });
10636:     }
10637:     const now = yield* DateTime.now;
10638:     // Record that Stop reached the turn it could not interrupt and the thread's last executed
10639:     // run, so a late tool call from their agent starts nothing.
10640:     const executed = latestExecutedRun(projection.runs);
10641:     const reached = new Map<RunId, OrchestrationV2Run>();
10642:     if (target !== undefined) reached.set(target.id, target);
10643:     if (executed !== null && !hasLiveRun({ runs: [executed] })) reached.set(executed.id, executed);
10644:     for (const run of reached.values()) {
10645:       yield* markStoppedRun({ command, events, thread: projection.thread, run, now });
10646:     }
10647:     yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now });
10648:     yield* cancelIdleOrphans({ command, events, projection });
10649:     yield* stopDelegatedTasks;
10650:     return undefined;
10651:   });
11925:         yield* dispatchRuntimeRequestRespond(command, events, effects);
11926:         break;
11927:       case "thread.user-input.dismiss":
11928:         yield* dispatchThreadUserInputDismiss(command, events, effects);
11929:         break;
11930:       case "subagent.stop":
11931:         yield* dispatchSubagentStop(command, events, effects);
11932:         break;
11933:       case "run.interrupt":
11934:         cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects);
11935:         if (command.holdQueue === true) {
11936:           // Stop must still commit when a child can't be read, so a failed repair's
11937:           // partial events are dropped.
11938:           const repairEvents = yield* Ref.make(yield* Ref.get(events));
11939:           const repaired = yield* Effect.exit(
11940:             loadProjectionForCommand(command, ["subagents", "turnItems"], {
11941:               turnItemTypes: ["subagent"],
11942:               turnItemStatuses: ["pending", "running", "waiting"],
11943:             }).pipe(
11944:               Effect.flatMap((projection) =>
11945:                 cancelIdleOrphans({ command, events: repairEvents, projection }),
11946:               ),
11947:             ),
11948:           );
11949:           if (Exit.isSuccess(repaired)) yield* Ref.set(events, yield* Ref.get(repairEvents));
11950:           else
11951:             yield* Effect.logWarning("Stop could not cancel orphaned delegated follow-ups", {
11952:               threadId: command.threadId,
11953:               cause: repaired.cause,
11954:             });
11955:           yield* Ref.update(effects, (existing) => [
11956:             ...existing,
11957:             delegatedTasksStopEffect(command),
11958:           ]);
11959:         }
11960:         break;
11961:       case "thread.stop":
11962:         cancelUnsettledEffects = yield* dispatchThreadStop(command, events, effects);
11963:         break;
11964:       case "queued-message.promote-to-steer":
11965:         cancelUnsettledEffects = yield* dispatchQueuedMessagePromoteToSteer(
```

## apps/server/src/orchestration-v2/ThreadManagementService.ts

Whole-file identical: True.

[96fc7b9551](https://git.cafe/pingdotgg/t3code/blob/96fc7b955196637605cbf3b415152ce4dd0e2294/apps/server/src/orchestration-v2/ThreadManagementService.ts), SHA256 `df659e939b24ab4737a4455d4df8f75161cd388016232c4c8f585a590729340e`

```ts
1165:   const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = Effect.fn(
1166:     "ThreadManagementService.stopDelegatedTasks",
1167:   )(function* (input) {
1168:     const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]);
1169:     const failures: Array<
1170:       Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
1171:     > = [];
1172:     const stoppedChildThreadIds = new Set<ThreadId>();
1173:     for (const task of subagents) {
1174:       if (task.origin !== "app_owned" || task.childThreadId === null) continue;
1175:       const threadId = task.childThreadId;
1176:       // Follow-up tasks reuse a child thread; walk its subtree only once.
1177:       if (stoppedChildThreadIds.has(threadId)) continue;
1178:       stoppedChildThreadIds.add(threadId);
1179:       yield* dispatch({
1180:         type: "thread.stop",
1181:         commandId: CommandId.make(`${input.commandId}:stop:${threadId}`),
1182:         threadId,
1183:         stopDelegatedTasks: false,
1184:         ...(input.reason === undefined ? {} : { reason: input.reason }),
1185:       }).pipe(
1186:         Effect.andThen(stopDelegatedTasks({ ...input, threadId })),
1187:         Effect.catch((error) =>
1188:           Effect.logWarning("Unable to stop a delegated task", {
1189:             parentThreadId: input.threadId,
1190:             threadId,
1191:             error,
1192:           }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))),
1193:         ),
1194:       );
1195:     }
1196:     if (failures[0] !== undefined) return yield* failures[0];
1197:   });
1198: 
1199:   return ThreadManagementService.of({
1200:     ensureLegacyTranscript,
1201:     searchThreadStream: (input) =>
1202:       Stream.unwrap(
1203:         ensureProjectionTranscript(input.threadId).pipe(
1204:           Effect.as(orchestrator.searchThreadStream(input)),
```

[802efb97da](https://git.cafe/pingdotgg/t3code/blob/802efb97da70e935d2c55ff2be6caf1093ffdb32/apps/server/src/orchestration-v2/ThreadManagementService.ts), SHA256 `df659e939b24ab4737a4455d4df8f75161cd388016232c4c8f585a590729340e`

```ts
1165:   const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = Effect.fn(
1166:     "ThreadManagementService.stopDelegatedTasks",
1167:   )(function* (input) {
1168:     const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]);
1169:     const failures: Array<
1170:       Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
1171:     > = [];
1172:     const stoppedChildThreadIds = new Set<ThreadId>();
1173:     for (const task of subagents) {
1174:       if (task.origin !== "app_owned" || task.childThreadId === null) continue;
1175:       const threadId = task.childThreadId;
1176:       // Follow-up tasks reuse a child thread; walk its subtree only once.
1177:       if (stoppedChildThreadIds.has(threadId)) continue;
1178:       stoppedChildThreadIds.add(threadId);
1179:       yield* dispatch({
1180:         type: "thread.stop",
1181:         commandId: CommandId.make(`${input.commandId}:stop:${threadId}`),
1182:         threadId,
1183:         stopDelegatedTasks: false,
1184:         ...(input.reason === undefined ? {} : { reason: input.reason }),
1185:       }).pipe(
1186:         Effect.andThen(stopDelegatedTasks({ ...input, threadId })),
1187:         Effect.catch((error) =>
1188:           Effect.logWarning("Unable to stop a delegated task", {
1189:             parentThreadId: input.threadId,
1190:             threadId,
1191:             error,
1192:           }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))),
1193:         ),
1194:       );
1195:     }
1196:     if (failures[0] !== undefined) return yield* failures[0];
1197:   });
1198: 
1199:   return ThreadManagementService.of({
1200:     ensureLegacyTranscript,
1201:     searchThreadStream: (input) =>
1202:       Stream.unwrap(
1203:         ensureProjectionTranscript(input.threadId).pipe(
1204:           Effect.as(orchestrator.searchThreadStream(input)),
```

## apps/server/src/mcp/OrchestratorMcpService.ts

Whole-file identical: True.

[96fc7b9551](https://git.cafe/pingdotgg/t3code/blob/96fc7b955196637605cbf3b415152ce4dd0e2294/apps/server/src/mcp/OrchestratorMcpService.ts), SHA256 `34209922d71be87c173019ac0f9682276ff267729aae81b5bdf6ab5686788d26`

```ts
2368:                   ),
2369:                 ),
2370:               );
2371:       const stopChild = Effect.gen(function* () {
2372:         const commandId = stableCommandId({ scope, requestKey: key, operation: "cancel-task" });
2373:         const reason = agentStopReason(input.reason);
2374:         yield* threadManagement
2375:           .dispatch({
2376:             type: "thread.stop",
2377:             commandId,
2378:             threadId: current.childThreadId,
2379:             stopDelegatedTasks: false,
2380:             reason,
2381:           })
2382:           .pipe(
2383:             Effect.mapError((error) =>
2384:               failure(
2385:                 "task_not_cancellable",
2386:                 `Unable to stop delegated task ${input.taskId}: ${errorMessage(error)}`,
2387:               ),
2388:             ),
2389:           );
2390:         yield* threadManagement
2391:           .stopDelegatedTasks({ threadId: current.childThreadId, commandId, reason })
2392:           .pipe(
2393:             Effect.mapError((error) =>
2394:               failure(
2395:                 "task_not_cancellable",
2396:                 `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`,
2397:               ),
2398:             ),
2399:           );
2400:         const stopped = yield* threadManagement
```

[802efb97da](https://git.cafe/pingdotgg/t3code/blob/802efb97da70e935d2c55ff2be6caf1093ffdb32/apps/server/src/mcp/OrchestratorMcpService.ts), SHA256 `34209922d71be87c173019ac0f9682276ff267729aae81b5bdf6ab5686788d26`

```ts
2368:                   ),
2369:                 ),
2370:               );
2371:       const stopChild = Effect.gen(function* () {
2372:         const commandId = stableCommandId({ scope, requestKey: key, operation: "cancel-task" });
2373:         const reason = agentStopReason(input.reason);
2374:         yield* threadManagement
2375:           .dispatch({
2376:             type: "thread.stop",
2377:             commandId,
2378:             threadId: current.childThreadId,
2379:             stopDelegatedTasks: false,
2380:             reason,
2381:           })
2382:           .pipe(
2383:             Effect.mapError((error) =>
2384:               failure(
2385:                 "task_not_cancellable",
2386:                 `Unable to stop delegated task ${input.taskId}: ${errorMessage(error)}`,
2387:               ),
2388:             ),
2389:           );
2390:         yield* threadManagement
2391:           .stopDelegatedTasks({ threadId: current.childThreadId, commandId, reason })
2392:           .pipe(
2393:             Effect.mapError((error) =>
2394:               failure(
2395:                 "task_not_cancellable",
2396:                 `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`,
2397:               ),
2398:             ),
2399:           );
2400:         const stopped = yield* threadManagement
```

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

Whole-file identical: False.

[96fc7b9551](https://git.cafe/pingdotgg/t3code/blob/96fc7b955196637605cbf3b415152ce4dd0e2294/apps/server/src/orchestration-v2/EffectWorker.ts), SHA256 `be94f863dc1ab855e4678c27c8f277effadc051b2cde52a0660872bea7071409`

```ts
          // Arm cancellation before rereading durable state to close the read-to-registration race.
          const cancellation = outbox
            .awaitCancellation(effect.id)
            .pipe(Effect.as("cancelled" as const));
          const cancelledBeforeExecution = yield* Effect.gen(function* () {
            const claimedAt = DateTime.toEpochMillis(yield* DateTime.now);
            const eligibleAt = Math.max(
              DateTime.toEpochMillis(DateTime.makeUnsafe(effect.createdAt)),
              DateTime.toEpochMillis(DateTime.makeUnsafe(effect.availableAt)),
            );
            yield* Metric.update(
              Metric.withAttributes(
                orchestrationEffectQueueWait,
                metricAttributes({ effect_type: effect.request.type }),
              ),
              Duration.millis(Math.max(0, claimedAt - eligibleAt)),
            );
            // Recheck durable cancellation after registering the Deferred; cancellation may have committed after claim.
            if (yield* wasCancelled(effect.id)) {
              yield* outbox.clearCancellation(effect.id);
              return true;
            }
            return false;
          }).pipe(Effect.onError((cause) => requeueClaim(effect, cause)));
          if (cancelledBeforeExecution) return true;


```

[802efb97da](https://git.cafe/pingdotgg/t3code/blob/802efb97da70e935d2c55ff2be6caf1093ffdb32/apps/server/src/orchestration-v2/EffectWorker.ts), SHA256 `29e2e4e43aa75d97fa29e575dfdaee3bc0b545265a7dd9fad6f7e3d0bf0d63ab`

```ts
          // Arm cancellation before rereading durable state to close the read-to-registration race.
          const cancellation = outbox
            .awaitCancellation(effect.id)
            .pipe(Effect.as("cancelled" as const));
          const cancelledBeforeExecution = yield* Effect.gen(function* () {
            const claimedAt = DateTime.toEpochMillis(yield* DateTime.now);
            const eligibleAt = Math.max(
              DateTime.toEpochMillis(DateTime.makeUnsafe(effect.createdAt)),
              DateTime.toEpochMillis(DateTime.makeUnsafe(effect.availableAt)),
            );
            yield* Metric.update(
              Metric.withAttributes(
                orchestrationEffectQueueWait,
                metricAttributes({ effect_type: effect.request.type }),
              ),
              Duration.millis(Math.max(0, claimedAt - eligibleAt)),
            );
            // Recheck durable cancellation after registering the Deferred; cancellation may have committed after claim.
            if (yield* wasCancelled(effect.id)) {
              yield* outbox.clearCancellation(effect.id);
              return true;
            }
            return false;
          }).pipe(Effect.onError((cause) => requeueClaim(effect, cause)));
          if (cancelledBeforeExecution) return true;


```

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

Whole-file identical: True.

[96fc7b9551](https://git.cafe/pingdotgg/t3code/blob/96fc7b955196637605cbf3b415152ce4dd0e2294/apps/server/src/orchestration-v2/EffectOutbox.ts), SHA256 `e1139c7a92c99c3629d63f3b681b472ff5451d3d475fd690728d37515d84f48d`

```ts
317:     // Retry backoff must block later effects so turns cannot race a pending rollback.
318:     const claimableCandidatePredicate = (
319:       availableBefore?: string,
320:       excludeRestartContinuations = false,
321:     ) =>
322:       sql`
323:         ${
324:           availableBefore === undefined
325:             ? sql`1 = 1`
326:             : sql`candidate.available_at <= ${availableBefore}`
327:         }
328:         AND candidate.status = 'pending'
329:         AND NOT EXISTS (
330:           SELECT 1
331:           FROM orchestration_v2_effect_outbox AS active
332:           WHERE active.thread_id = candidate.thread_id
333:             AND (
334:               active.status = 'running'
335:               OR (
336:                 active.status = 'pending'
337:                 AND active.rowid < candidate.rowid
338:                 AND ${
339:                   excludeRestartContinuations
340:                     ? sql`active.effect_type != 'provider-runtime.continue'`
341:                     : sql`1 = 1`
342:                 }
343:               )
344:             )
345:             AND (
346:               (
347:                 candidate.effect_type = 'thread-title.generate'
348:                 AND active.effect_type = 'thread-title.generate'
349:               )
350:               OR
351:               (
352:                 candidate.effect_type != 'thread-title.generate'
353:                 AND active.effect_type != 'thread-title.generate'
354:               )
355:             )
356:         )
357:       `;
358: 
359:     const cancellationSignal = (effectId: string) => {
360:       const existing = cancellationSignals.get(effectId);
```

[802efb97da](https://git.cafe/pingdotgg/t3code/blob/802efb97da70e935d2c55ff2be6caf1093ffdb32/apps/server/src/orchestration-v2/EffectOutbox.ts), SHA256 `e1139c7a92c99c3629d63f3b681b472ff5451d3d475fd690728d37515d84f48d`

```ts
317:     // Retry backoff must block later effects so turns cannot race a pending rollback.
318:     const claimableCandidatePredicate = (
319:       availableBefore?: string,
320:       excludeRestartContinuations = false,
321:     ) =>
322:       sql`
323:         ${
324:           availableBefore === undefined
325:             ? sql`1 = 1`
326:             : sql`candidate.available_at <= ${availableBefore}`
327:         }
328:         AND candidate.status = 'pending'
329:         AND NOT EXISTS (
330:           SELECT 1
331:           FROM orchestration_v2_effect_outbox AS active
332:           WHERE active.thread_id = candidate.thread_id
333:             AND (
334:               active.status = 'running'
335:               OR (
336:                 active.status = 'pending'
337:                 AND active.rowid < candidate.rowid
338:                 AND ${
339:                   excludeRestartContinuations
340:                     ? sql`active.effect_type != 'provider-runtime.continue'`
341:                     : sql`1 = 1`
342:                 }
343:               )
344:             )
345:             AND (
346:               (
347:                 candidate.effect_type = 'thread-title.generate'
348:                 AND active.effect_type = 'thread-title.generate'
349:               )
350:               OR
351:               (
352:                 candidate.effect_type != 'thread-title.generate'
353:                 AND active.effect_type != 'thread-title.generate'
354:               )
355:             )
356:         )
357:       `;
358: 
359:     const cancellationSignal = (effectId: string) => {
360:       const existing = cancellationSignals.get(effectId);
```