1import { modelSelectionsEqual } from "@t3tools/shared/model";
2import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences";
3import {
4 CommandId,
5 latestProviderTurnForAttempt,
6 type OrchestrationV2DomainEvent,
7 type OrchestrationV2ExecutionNode,
8 type OrchestrationV2ProviderThread,
9 type OrchestrationV2Run,
10 type OrchestrationV2RunAttempt,
11 type OrchestrationV2TurnItem,
12 RunId,
13 ThreadId,
14} from "@t3tools/contracts";
15import * as Context from "effect/Context";
16import * as Cause from "effect/Cause";
17import * as DateTime from "effect/DateTime";
18import * as Effect from "effect/Effect";
19import * as FileSystem from "effect/FileSystem";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Path from "effect/Path";
23import * as Predicate from "effect/Predicate";
24import * as Schema from "effect/Schema";
25
26import { withWorkspaceLease } from "../workspace/workspaceLease.ts";
27import * as GitWorkflowService from "../git/GitWorkflowService.ts";
28import * as ProjectService from "../project/ProjectService.ts";
29import * as ProviderAuthService from "../provider/ProviderAuthService.ts";
30import * as EventSink from "./EventSink.ts";
31import * as ContextHandoffService from "./ContextHandoffService.ts";
32import {
33 DEFAULT_HANDOFF_TOKEN_CAP,
34 handoffTokenCapConfig,
35 handoffBudget,
36 attachmentTokenAllowance,
37 contextUsageForHandoff,
38 historicalMessage,
39 latestNativeContextUsage,
40} from "@t3tools/provider-core/server/handoffBudget";
41import { deliverContextHandoffs } from "./ContextHandoffDelivery.ts";
42import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
43import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
44import * as ProjectionStore from "./ProjectionStore.ts";
45import * as ProviderSessionManager from "./ProviderSessionManager.ts";
46import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
47import * as RunExecutionService from "./RunExecutionService.ts";
48import * as RuntimePolicy from "./RuntimePolicy.ts";
49import {
50 isRestartNoteContinuation,
51 pendingRestartCancelledBackgroundWork,
52 restartCancelledBackgroundWorkNote,
53} from "./RestartBackgroundNote.ts";
54
55export class ProviderTurnStartError extends Schema.TaggedError<ProviderTurnStartError>()(
56 "ProviderTurnStartError",
57 {
58 runId: RunId,
59 cause: Schema.optional(Schema.Defect()),
60 },
61) {}
62
63const isProviderTurnStartError = Schema.is(ProviderTurnStartError);
64
65/** Claude refuses to replace a process running background work before it reads the prompt. */
66const refusedBeforePrompt = (error: unknown): boolean =>
67 Predicate.isTagged(error, "ClaudeBackgroundWorkBlocksQueryReplacementError") ||
68 (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause));
69
70export interface ProviderTurnStartServiceV2Shape {
71 /**
72 * Starts the run's provider turn. When `willRetry` is true, a session open
73 * failure is returned so the caller can retry. Otherwise the run is settled
74 * as failed.
75 */
76 readonly start: (input: {
77 readonly threadId: ThreadId;
78 readonly runId: RunId;
79 readonly willRetry?: boolean;
80 }) => Effect.Effect<void, ProviderTurnStartError>;
81}
82
83export class ProviderTurnStartServiceV2 extends Context.Service<
84 ProviderTurnStartServiceV2,
85 ProviderTurnStartServiceV2Shape
86>()("t3/orchestration-v2/ProviderTurnStartService/ProviderTurnStartServiceV2") {}
87
88export const layer: Layer.Layer<
89 ProviderTurnStartServiceV2,
90 never,
91 | EventSink.EventSinkV2
92 | ContextHandoffService.ContextHandoffServiceV2
93 | IdAllocator.IdAllocatorV2
94 | FileSystem.FileSystem
95 | Path.Path
96 | GitWorkflowService.GitWorkflowService
97 | ProjectService.ProjectService
98 | ProviderAuthService.ProviderAuthService
99 | ProjectionStore.ProjectionStoreV2
100 | ProviderSessionManager.ProviderSessionManagerV2
101 | RunExecutionService.RunExecutionServiceV2
102 | RuntimePolicy.RuntimePolicyV2
103> = Layer.effect(
104 ProviderTurnStartServiceV2,
105 Effect.gen(function* () {
106 const eventSink = yield* EventSink.EventSinkV2;
107 const contextHandoffService = yield* ContextHandoffService.ContextHandoffServiceV2;
108 const idAllocator = yield* IdAllocator.IdAllocatorV2;
109 const fileSystem = yield* FileSystem.FileSystem;
110 const path = yield* Path.Path;
111 const gitWorkflow = yield* GitWorkflowService.GitWorkflowService;
112 const projects = yield* ProjectService.ProjectService;
113 const providerAuth = yield* ProviderAuthService.ProviderAuthService;
114 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
115 const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
116 const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
117 const runtimePolicy = yield* RuntimePolicy.RuntimePolicyV2;
118
119 // These callbacks outlive startup while a run drains background work. Build
120 // them outside start's scope so they cannot retain its full thread history.
121 const makeRunControls = (input: {
122 readonly threadId: ThreadId;
123 readonly runId: RunId;
124 readonly attemptId: OrchestrationV2RunAttempt["id"];
125 readonly providerThreadId: OrchestrationV2ProviderThread["id"];
126 readonly runOrdinal: number;
127 readonly inheritedBackgroundTurnItems: ReturnType<
128 typeof RunExecutionService.selectInheritedBackgroundTurnItems
129 >;
130 }) => {
131 // Guards and background routing need live execution state, not a fresh
132 // allocation of every completed message and tool output in the thread.
133 // `false` means the run moved on or is gone. A failed read is an error,
134 // so the caller fails the start or the run instead of skipping it.
135 const isCurrentAttemptInStatus = (expectedStatus: OrchestrationV2Run["status"]) =>
136 projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe(
137 Effect.map((current) => {
138 const run = current.runs.find((candidate) => candidate.id === input.runId);
139 return run?.activeAttemptId === input.attemptId && run.status === expectedStatus;
140 }),
141 );
142 return {
143 isCurrentAttemptInStatus,
144 loadInheritedBackgroundTurnItems: () =>
145 projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe(
146 Effect.map((current) =>
147 RunExecutionService.selectInheritedBackgroundTurnItems({
148 threadId: input.threadId,
149 currentProviderThreadId: input.providerThreadId,
150 currentRunOrdinal: input.runOrdinal,
151 runs: current.runs,
152 turnItems: current.turnItems,
153 }),
154 ),
155 Effect.catchCause(() => Effect.succeed(input.inheritedBackgroundTurnItems)),
156 ),
157 shouldStartProviderTurn: () => isCurrentAttemptInStatus("running"),
158 shouldFinalizeRun: () =>
159 projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe(
160 Effect.map((current) => {
161 const run = current.runs.find((candidate) => candidate.id === input.runId);
162 return (
163 run?.activeAttemptId === input.attemptId &&
164 (run.status === "starting" || run.status === "running")
165 );
166 }),
167 ),
168 hasUnpairedRunInterruptRequest: () =>
169 projectionStore
170 .hasUnpairedRunInterruptRequest(
171 input.threadId,
172 idAllocator.derive.runSignalTurnItem({
173 runId: input.runId,
174 signal: "interrupt-request",
175 }),
176 idAllocator.derive.runSignalTurnItem({
177 runId: input.runId,
178 signal: "interrupt-result",
179 }),
180 )
181 .pipe(Effect.catchCause(() => Effect.succeed(false))),
182 };
183 };
184
185 const makeDeliverySession = (
186 session: ProviderAdapter.ProviderAdapterV2SessionRuntime,
187 startWithHandoffs: (
188 input: Parameters<ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"]>[0],
189 compact?: boolean,
190 ) => ReturnType<ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"]>,
191 ) => {
192 let deliver: typeof startWithHandoffs | undefined = startWithHandoffs;
193 const start = (
194 input: Parameters<ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"]>[0],
195 compact = false,
196 ) =>
197 Effect.suspend(() => {
198 if (deliver !== undefined) return deliver(input, compact);
199 return compact && session.compactThread !== undefined
200 ? session.compactThread(input)
201 : session.startTurn(input);
202 }).pipe(
203 // Only startup needs the handoff history. The event worker keeps this
204 // session alive afterward, including when background work remains.
205 Effect.ensuring(
206 Effect.sync(() => {
207 deliver = undefined;
208 }),
209 ),
210 );
211 return {
212 ...session,
213 startTurn: (input: Parameters<typeof session.startTurn>[0]) => start(input),
214 ...(session.compactThread === undefined
215 ? {}
216 : {
217 compactThread: (input: Parameters<typeof session.startTurn>[0]) => start(input, true),
218 }),
219 };
220 };
221
222 const start = Effect.fn("orchestrationV2.providerTurnStart.start")(function* (input: {
223 readonly threadId: ThreadId;
224 readonly runId: RunId;
225 readonly willRetry?: boolean;
226 }) {
227 const { runId } = input;
228 const projection = yield* projectionStore.getTurnStartContext(input.threadId, runId);
229 const run = projection.runs.find((candidate) => candidate.id === runId);
230 if (run === undefined) {
231 return yield* new ProviderTurnStartError({ runId, cause: `Run ${runId} was not found.` });
232 }
233 if (run.status !== "starting") {
234 // The effect is idempotent once the run has advanced or terminalized.
235 return;
236 }
237 const rootNode = projection.nodes.find((candidate) => candidate.id === run.rootNodeId);
238 const attempt = projection.attempts.find((candidate) => candidate.id === run.activeAttemptId);
239 const providerThread = projection.providerThreads.find(
240 (candidate) => candidate.id === run.providerThreadId,
241 );
242 const message = projection.messages.find((candidate) => candidate.id === run.userMessageId);
243 const checkpointScope = projection.checkpointScopes.find(
244 (candidate) => candidate.id === rootNode?.checkpointScopeId,
245 );
246 const handoffs = projection.contextHandoffs.filter(
247 (handoff) =>
248 handoff.status === "ready" &&
249 (handoff.targetRunId === run.id ||
250 (handoff.toProviderThreadId === run.providerThreadId &&
251 projection.runs.some(
252 (source) =>
253 source.id === handoff.targetRunId &&
254 (source.status === "failed" ||
255 source.status === "interrupted" ||
256 (source.status === "completed" &&
257 handoff.delivery === undefined &&
258 projection.messages.some(
259 (message) =>
260 message.id === source.userMessageId &&
261 message.attachments.length === 0 &&
262 message.text.trim().toLowerCase() === "/compact",
263 ))),
264 ))),
265 );
266 const nativeForkTransfer = projection.contextTransfers.find(
267 (transfer) =>
268 transfer.type === "fork" &&
269 transfer.targetThreadId === input.threadId &&
270 transfer.targetRunId === run.id &&
271 transfer.status === "pending" &&
272 transfer.resolution === null,
273 );
274 if (
275 rootNode === undefined ||
276 attempt === undefined ||
277 providerThread === undefined ||
278 providerThread.providerSessionId === null ||
279 message === undefined ||
280 checkpointScope === undefined
281 ) {
282 return yield* new ProviderTurnStartError({
283 runId,
284 cause: `Run ${runId} is missing its execution projection state.`,
285 });
286 }
287 // Settles a run that never reached the provider: one signal turn item plus
288 // terminal run, attempt and root node, written only while the run is still
289 // the current starting attempt.
290 const settleRunBeforeStart = Effect.fn("orchestrationV2.providerTurnStart.settleBeforeStart")(
291 function* (input: {
292 readonly signal: string;
293 readonly status: "completed" | "failed";
294 readonly now: DateTime.Utc;
295 /** Omitted when the run never started, so `startedAt` stays as projected. */
296 readonly startedAt?: DateTime.Utc;
297 readonly providerInstanceId: OrchestrationV2Run["providerInstanceId"];
298 readonly itemProviderThreadId: OrchestrationV2ProviderThread["id"];
299 readonly item:
300 | Pick<
301 Extract<OrchestrationV2TurnItem, { type: "error" }>,
302 "type" | "title" | "failure"
303 >
304 | Pick<
305 Extract<OrchestrationV2TurnItem, { type: "command_execution" }>,
306 "type" | "title" | "input" | "output" | "exitCode"
307 >;
308 /** Emitted after the run events when the provider thread should go idle. */
309 readonly providerThreadUpdate?: OrchestrationV2ProviderThread;
310 }) {
311 const { now, status } = input;
312 const started = input.startedAt === undefined ? {} : { startedAt: input.startedAt };
313 const item: OrchestrationV2TurnItem = {
314 id: idAllocator.derive.runSignalTurnItem({ runId, signal: input.signal }),
315 threadId: projection.thread.id,
316 runId,
317 nodeId: rootNode.id,
318 providerThreadId: input.itemProviderThreadId,
319 providerTurnId: null,
320 nativeItemRef: null,
321 parentItemId: null,
322 ordinal:
323 Math.max(
324 0,
325 ...projection.turnItems
326 .filter((item) => item.runId === runId)
327 .map((item) => item.ordinal),
328 ) + 1,
329 status,
330 startedAt: now,
331 completedAt: now,
332 updatedAt: now,
333 ...input.item,
334 };
335 const eventPayloads = [
336 { type: "turn-item.updated", payload: item },
337 { type: "run.updated", payload: { ...run, status, ...started, completedAt: now } },
338 {
339 type: "run-attempt.updated",
340 payload: { ...attempt, status, ...started, completedAt: now },
341 },
342 {
343 type: "node.updated",
344 payload: { ...rootNode, status, ...started, completedAt: now },
345 },
346 ...(input.providerThreadUpdate === undefined
347 ? []
348 : [
349 {
350 type: "provider-thread.updated" as const,
351 payload: input.providerThreadUpdate,
352 },
353 ]),
354 ] as const;
355 const events = yield* Effect.forEach(eventPayloads, (event) =>
356 Effect.gen(function* () {
357 return {
358 ...event,
359 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
360 threadId: projection.thread.id,
361 runId,
362 nodeId: rootNode.id,
363 providerInstanceId: input.providerInstanceId,
364 occurredAt: now,
365 } satisfies OrchestrationV2DomainEvent;
366 }),
367 );
368 yield* eventSink.writeIfRunCurrent({
369 threadId: projection.thread.id,
370 runId,
371 activeAttemptId: attempt.id,
372 expectedStatus: "starting",
373 events,
374 });
375 },
376 );
377 if (message.attachments.length === 0 && message.text.trimStart().startsWith("/")) {
378 const isEmptyCompaction =
379 message.text.trim().toLowerCase() === "/compact" && !projection.hasConversation;
380 // Preparing a run may already point the thread at a newly selected
381 // provider. Account commands still belong to its last native session.
382 const nativeThreads = new Map(
383 projection.providerThreads
384 .filter(
385 (candidate) => candidate.ownerNodeId === null && candidate.nativeThreadRef !== null,
386 )
387 .map((candidate) => [candidate.id, candidate]),
388 );
389 const previousNativeRun = projection.runs.reduce<OrchestrationV2Run | undefined>(
390 (previous, candidate) =>
391 candidate.ordinal < run.ordinal &&
392 candidate.providerThreadId !== null &&
393 nativeThreads.has(candidate.providerThreadId) &&
394 (previous === undefined || candidate.ordinal > previous.ordinal)
395 ? candidate
396 : previous,
397 undefined,
398 );
399 const nativeThread = nativeThreads.get(
400 previousNativeRun?.providerThreadId ??
401 projection.thread.activeProviderThreadId ??
402 providerThread.id,
403 );
404 const authInstanceId = nativeThread?.providerInstanceId ?? run.providerInstanceId;
405 const authResult = isEmptyCompaction
406 ? null
407 : yield* Effect.result(
408 providerAuth.tryHandlePromptCommand({
409 instanceId: authInstanceId,
410 text: projectComposerContextForProvider({
411 text: message.text,
412 records: message.context?.records ?? [],
413 }),
414 hasAttachments: false,
415 }),
416 );
417 if (isEmptyCompaction || authResult?._tag === "Failure" || authResult?.success) {
418 const now = yield* DateTime.now;
419 const failure = isEmptyCompaction
420 ? makeProviderFailure({
421 class: "validation_error",
422 message: "Start a conversation before compacting this thread.",
423 })
424 : authResult?._tag === "Failure"
425 ? makeProviderFailure({
426 class: "permission_error",
427 message: authResult.failure.detail,
428 })
429 : undefined;
430 const status = failure === undefined ? "completed" : "failed";
431 yield* settleRunBeforeStart({
432 signal: isEmptyCompaction ? "empty-compaction" : "provider-sign-out",
433 status,
434 now,
435 startedAt: now,
436 providerInstanceId: authInstanceId,
437 itemProviderThreadId: nativeThread?.id ?? providerThread.id,
438 item:
439 failure !== undefined
440 ? {
441 type: "error",
442 title: isEmptyCompaction
443 ? "Cannot compact an empty thread"
444 : "Provider sign-out failed",
445 failure,
446 }
447 : {
448 type: "command_execution",
449 title: "Provider signed out",
450 input: message.text.trim(),
451 output: "Provider signed out",
452 exitCode: 0,
453 },
454 providerThreadUpdate: {
455 ...providerThread,
456 status: providerThread.nativeThreadRef === null ? "not_loaded" : "idle",
457 updatedAt: now,
458 },
459 });
460 return;
461 }
462 }
463 const { worktreePath, branch } = projection.thread;
464 if (worktreePath !== null && branch !== null) {
465 const exists = yield* fileSystem
466 .exists(worktreePath)
467 .pipe(Effect.orElseSucceed(() => true));
468 if (!exists) {
469 const project = yield* projects.getById(projection.thread.projectId).pipe(
470 Effect.map(Option.getOrUndefined),
471 Effect.orElseSucceed(() => undefined),
472 );
473 if (project !== undefined) {
474 yield* Effect.logWarning("provider turn start recreating missing worktree", {
475 threadId: projection.thread.id,
476 worktreePath,
477 branch,
478 });
479 yield* gitWorkflow.pruneWorktrees({ cwd: project.workspaceRoot }).pipe(
480 Effect.andThen(
481 gitWorkflow.createWorktree({
482 cwd: project.workspaceRoot,
483 refName: branch,
484 path: worktreePath,
485 }),
486 ),
487 Effect.catchCause((cause) =>
488 Cause.hasInterruptsOnly(cause)
489 ? Effect.failCause(cause)
490 : Effect.logWarning("provider turn start failed to recreate worktree", {
491 threadId: projection.thread.id,
492 worktreePath,
493 cause: Cause.pretty(cause),
494 }),
495 ),
496 );
497 }
498 }
499 }
500 const selectInheritedBackgroundItems = (
501 current: ProjectionStore.ProjectionRuntimeRecoveryState,
502 ): ReturnType<typeof RunExecutionService.selectInheritedBackgroundTurnItems> =>
503 RunExecutionService.selectInheritedBackgroundTurnItems({
504 threadId: current.thread.id,
505 currentProviderThreadId: providerThread.id,
506 currentRunOrdinal: run.ordinal,
507 runs: current.runs,
508 turnItems: current.turnItems,
509 });
510 const inheritedBackgroundTurnItems = yield* projectionStore
511 .getRuntimeRecoveryProjection(projection.thread.id)
512 .pipe(Effect.map(selectInheritedBackgroundItems));
513 const providerSessionId = providerThread.providerSessionId;
514 const runControls = makeRunControls({
515 threadId: projection.thread.id,
516 runId: run.id,
517 attemptId: attempt.id,
518 providerThreadId: providerThread.id,
519 runOrdinal: run.ordinal,
520 inheritedBackgroundTurnItems,
521 });
522 const { isCurrentAttemptInStatus } = runControls;
523
524 const resolvedRuntimePolicy = yield* runtimePolicy.resolve({
525 thread: projection.thread,
526 modelSelection: run.modelSelection,
527 });
528 const existingSessionProjection = projection.providerSessions.find(
529 (candidate) => candidate.id === providerSessionId,
530 );
531 const sessionResult = yield* Effect.result(
532 providerSessions.open({
533 threadId: projection.thread.id,
534 providerSessionId,
535 modelSelection: run.modelSelection,
536 runtimePolicy: resolvedRuntimePolicy,
537 ...(existingSessionProjection === undefined
538 ? {}
539 : { resumeFromSession: existingSessionProjection }),
540 ...(providerThread.nativeThreadRef?.nativeId == null
541 ? {}
542 : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }),
543 ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined
544 ? {}
545 : {
546 initialProviderItemIdentityVersion:
547 providerThread.nativeMetadata.itemIdentityVersion,
548 }),
549 }),
550 );
551 // The last start attempt fails the run with the provider's own reason
552 // instead of leaving it `starting` after the effect gives up. A run that
553 // already left `starting` is not overwritten, and a failed write returns
554 // its error to the effect worker.
555 const settleStartFailure = (failed: {
556 readonly signal: string;
557 readonly title: string;
558 readonly error: Error;
559 }) =>
560 Effect.gen(function* () {
561 const nestedCause = "cause" in failed.error ? failed.error.cause : undefined;
562 yield* settleRunBeforeStart({
563 signal: failed.signal,
564 status: "failed",
565 now: yield* DateTime.now,
566 providerInstanceId: run.providerInstanceId,
567 itemProviderThreadId: providerThread.id,
568 item: {
569 type: "error",
570 title: failed.title,
571 failure: makeProviderFailure({
572 cause: failed.error,
573 message:
574 nestedCause instanceof Error
575 ? nestedCause.message
576 : typeof nestedCause === "string"
577 ? nestedCause
578 : failed.error.message,
579 class: "provider_error",
580 }),
581 },
582 });
583 });
584 if (sessionResult._tag === "Failure") {
585 if (input.willRetry === true) return yield* sessionResult.failure;
586 yield* settleStartFailure({
587 signal: "provider-session-open-failure",
588 title: "Provider session failed to open",
589 error: sessionResult.failure,
590 });
591 return;
592 }
593 const session = sessionResult.success;
594 // Only the provider's own thread load fails the run on the last attempt;
595 // store, id and handoff failures around it keep their typed errors.
596 const loadFromProvider = (
597 load: Effect.Effect<OrchestrationV2ProviderThread, ProviderAdapter.ProviderAdapterV2Error>,
598 ) =>
599 Effect.gen(function* () {
600 const loaded = yield* Effect.result(load);
601 if (loaded._tag === "Success") return loaded.success;
602 if (input.willRetry === true) return yield* loaded.failure;
603 yield* settleStartFailure({
604 signal: "provider-thread-load-failure",
605 title: "Provider turn failed to start",
606 error: loaded.failure,
607 });
608 return undefined;
609 });
610 let effectiveHandoffs = handoffs;
611 const loadedProviderThread = yield* Effect.gen(function* () {
612 if (nativeForkTransfer !== undefined) {
613 const sourceProjection = yield* projectionStore.getThreadRecords(
614 nativeForkTransfer.sourceThreadId,
615 ["runs", "providerThreads", "attempts", "providerTurns"],
616 );
617 const sourceRun = sourceProjection.runs.find(
618 (candidate) => candidate.id === nativeForkTransfer.sourcePoint.runId,
619 );
620 const sourceProviderThread = sourceProjection.providerThreads.find(
621 (candidate) => candidate.id === sourceRun?.providerThreadId,
622 );
623 const sourceAttempt = sourceProjection.attempts.find(
624 (candidate) => candidate.id === sourceRun?.activeAttemptId,
625 );
626 const sourceProviderTurn =
627 latestProviderTurnForAttempt(sourceProjection.providerTurns, sourceAttempt?.id) ??
628 sourceProjection.providerTurns.find(
629 (candidate) => candidate.id === sourceAttempt?.providerTurnId,
630 );
631 if (sourceRun === undefined || sourceProviderThread === undefined) {
632 return yield* new ProviderTurnStartError({
633 runId,
634 cause: `Native fork transfer ${nativeForkTransfer.id} has no source provider execution.`,
635 });
636 }
637 return yield* loadFromProvider(
638 session.forkThread({
639 sourceProviderThread,
640 sourceProviderTurns: sourceProjection.providerTurns,
641 targetThreadId: projection.thread.id,
642 modelSelection: run.modelSelection,
643 runtimePolicy: resolvedRuntimePolicy,
644 ...(sourceProviderTurn === undefined
645 ? {}
646 : { providerTurnId: sourceProviderTurn.id }),
647 }),
648 );
649 }
650 if (providerThread.nativeThreadRef === null) {
651 // Hand the run's provider thread to the adapter so it adopts this
652 // row's identity when attaching native state. An adapter that mints
653 // its own row instead leaves two live rows per app thread, and
654 // `activeProviderThreadId` then flaps between them on every update.
655 return yield* loadFromProvider(
656 session.ensureThread({
657 threadId: projection.thread.id,
658 title: projection.thread.title,
659 modelSelection: run.modelSelection,
660 runtimePolicy: resolvedRuntimePolicy,
661 providerSessionId,
662 existingProviderThread: providerThread,
663 }),
664 );
665 }
666 const uncertainDelivery = projection.contextHandoffs.some(
667 (handoff) =>
668 handoff.toProviderThreadId === providerThread.id &&
669 handoff.delivery?.nativeThreadId === providerThread.nativeThreadRef?.nativeId &&
670 handoff.delivery?.status === "pending",
671 );
672 const resumed = yield* Effect.result(
673 uncertainDelivery
674 ? Effect.fail(
675 new ProviderAdapter.ProviderAdapterTurnStartError({
676 driver: session.driver,
677 threadId: projection.thread.id,
678 providerThreadId: providerThread.id,
679 runId,
680 cause: "Uncertain native history injection",
681 }),
682 )
683 : session.resumeThread({
684 providerThread,
685 threadId: projection.thread.id,
686 modelSelection: run.modelSelection,
687 runtimePolicy: resolvedRuntimePolicy,
688 }),
689 );
690 if (resumed._tag === "Success") {
691 return resumed.success;
692 }
693
694 yield* Effect.logWarning("Provider resume failed; attempting a fresh native session", {
695 driver: session.driver,
696 providerThreadId: providerThread.id,
697 runId,
698 reason: uncertainDelivery ? "uncertain_history_delivery" : "resume_failed",
699 errorTag: resumed.failure._tag,
700 });
701 const replacement = yield* loadFromProvider(
702 session.ensureThread({
703 threadId: projection.thread.id,
704 title: projection.thread.title,
705 modelSelection: run.modelSelection,
706 runtimePolicy: resolvedRuntimePolicy,
707 providerSessionId,
708 // The native ref is dropped so the adapter binds a fresh native
709 // session instead of retrying the resume that just failed, while
710 // still adopting this row's identity.
711 existingProviderThread: { ...providerThread, nativeThreadRef: null },
712 }),
713 );
714 if (replacement === undefined) return undefined;
715 const transferId = yield* idAllocator.allocate.contextTransfer({
716 sourceThreadId: projection.thread.id,
717 targetThreadId: projection.thread.id,
718 type: "provider_resume_fallback",
719 });
720 const createdAt = yield* DateTime.now;
721 const handoff = yield* contextHandoffService.prepareProviderHandoff({
722 threadId: projection.thread.id,
723 targetRunId: run.id,
724 transferId,
725 fromProviderThreadIds: [providerThread.id],
726 toProviderThreadId: providerThread.id,
727 fromProviderInstanceId: providerThread.providerInstanceId,
728 toProviderInstanceId: run.providerInstanceId,
729 coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) },
730 strategy: "full_thread_summary",
731 runs: projection.runs,
732 items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter(
733 (item) =>
734 item.runId === null ||
735 projection.runs.some(
736 (source) => source.id === item.runId && source.ordinal < run.ordinal,
737 ),
738 ),
739 createdAt,
740 });
741 effectiveHandoffs = [handoff, ...effectiveHandoffs];
742 yield* eventSink.write({
743 events: [
744 {
745 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
746 type: "context-handoff.updated",
747 threadId: projection.thread.id,
748 runId: run.id,
749 providerInstanceId: run.providerInstanceId,
750 occurredAt: createdAt,
751 payload: handoff,
752 },
753 {
754 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
755 type: "context-transfer.updated",
756 threadId: projection.thread.id,
757 runId: run.id,
758 providerInstanceId: run.providerInstanceId,
759 occurredAt: createdAt,
760 payload: {
761 id: transferId,
762 type: "provider_handoff",
763 sourceThreadId: projection.thread.id,
764 targetThreadId: projection.thread.id,
765 sourcePoint: { threadId: projection.thread.id },
766 basePoint: null,
767 sourceProviderInstanceId: providerThread.providerInstanceId,
768 targetProviderInstanceId: run.providerInstanceId,
769 targetRunId: run.id,
770 status: "resolved_portable",
771 resolution: { strategy: "portable_context", contextHandoffId: handoff.id },
772 createdBy: "system",
773 error: null,
774 createdAt,
775 updatedAt: createdAt,
776 consumedAt: null,
777 },
778 },
779 ],
780 });
781 return replacement;
782 });
783 // The last attempt already failed the run.
784 if (loadedProviderThread === undefined) return;
785 if (!(yield* isCurrentAttemptInStatus("starting"))) {
786 return;
787 }
788 const now = yield* DateTime.now;
789 // Only started runs reached the provider-thread update below. Queued runs and
790 // failures during session setup cannot establish a new telemetry selection.
791 const measuredContext = latestNativeContextUsage(projection, providerThread);
792 const previousSelection =
793 measuredContext?.modelSelection ??
794 projection.runs.findLast(
795 (source) =>
796 source.ordinal < run.ordinal &&
797 source.startedAt !== null &&
798 source.providerThreadId === providerThread.id,
799 )?.modelSelection;
800 const sameSelection =
801 previousSelection === undefined ||
802 modelSelectionsEqual(previousSelection, run.modelSelection);
803 const sameNativeThread =
804 loadedProviderThread.nativeThreadRef?.nativeId === providerThread.nativeThreadRef?.nativeId;
805 const threadUsage = loadedProviderThread.contextUsage ?? providerThread.contextUsage;
806 const previousUsage = measuredContext
807 ? { ...threadUsage, ...measuredContext.usage }
808 : threadUsage;
809 const reuseTelemetry =
810 sameSelection ||
811 (previousSelection !== undefined &&
812 session.canReuseContextUsage?.(previousSelection, run.modelSelection) === true);
813 const knownModelWindow = session.getModelContextWindow?.(
814 run.modelSelection,
815 resolvedRuntimePolicy.cwd,
816 );
817 // Persist before delivery. Keep this native transcript's measured
818 // occupancy. A different model drops compaction telemetry and uses the
819 // new window when that window is known.
820 const handoffUsage = contextUsageForHandoff({
821 sameNativeThread,
822 sameSelection,
823 reuseTelemetry,
824 previousUsage,
825 knownModelWindow,
826 });
827 const runningProviderThread: OrchestrationV2ProviderThread = {
828 ...loadedProviderThread,
829 contextUsage: handoffUsage,
830 id: providerThread.id,
831 driver: session.driver,
832 providerInstanceId: run.providerInstanceId,
833 providerSessionId,
834 appThreadId: projection.thread.id,
835 ownerNodeId: providerThread.ownerNodeId,
836 firstRunOrdinal: providerThread.firstRunOrdinal ?? run.ordinal,
837 lastRunOrdinal: run.ordinal,
838 handoffIds: providerThread.handoffIds,
839 forkedFrom: providerThread.forkedFrom,
840 status: "active",
841 createdAt: providerThread.createdAt,
842 updatedAt: now,
843 };
844 const runningRun: OrchestrationV2Run = {
845 ...run,
846 status: "running",
847 startedAt: now,
848 };
849 const runningAttempt: OrchestrationV2RunAttempt = {
850 ...attempt,
851 ...(runningProviderThread.nativeThreadRef?.nativeId == null
852 ? {}
853 : { nativeThreadId: runningProviderThread.nativeThreadRef.nativeId }),
854 status: "running",
855 startedAt: now,
856 };
857 const runningRootNode: OrchestrationV2ExecutionNode = {
858 ...rootNode,
859 status: "running",
860 startedAt: now,
861 };
862 const events: Array<OrchestrationV2DomainEvent> = [
863 {
864 id: yield* idAllocator.allocate.event({
865 threadId: projection.thread.id,
866 providerSessionId,
867 }),
868 type: "provider-session.updated",
869 threadId: projection.thread.id,
870 driver: session.driver,
871 providerInstanceId: run.providerInstanceId,
872 occurredAt: now,
873 payload: session.providerSession,
874 },
875 {
876 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
877 type: "provider-thread.updated",
878 threadId: projection.thread.id,
879 driver: session.driver,
880 providerInstanceId: run.providerInstanceId,
881 occurredAt: now,
882 payload: runningProviderThread,
883 },
884 ...(nativeForkTransfer === undefined || runningProviderThread.nativeThreadRef === null
885 ? []
886 : [
887 {
888 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
889 type: "context-transfer.updated" as const,
890 threadId: projection.thread.id,
891 runId: run.id,
892 driver: session.driver,
893 providerInstanceId: run.providerInstanceId,
894 occurredAt: now,
895 payload: {
896 ...nativeForkTransfer,
897 targetProviderInstanceId: run.providerInstanceId,
898 targetRunId: run.id,
899 status: "consumed" as const,
900 resolution: {
901 strategy: "native_fork" as const,
902 providerThreadRef: runningProviderThread.nativeThreadRef,
903 },
904 error: null,
905 updatedAt: now,
906 consumedAt: now,
907 },
908 },
909 ]),
910 {
911 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
912 type: "run.updated",
913 threadId: projection.thread.id,
914 runId: run.id,
915 nodeId: rootNode.id,
916 providerInstanceId: run.providerInstanceId,
917 occurredAt: now,
918 payload: runningRun,
919 },
920 {
921 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
922 type: "run-attempt.updated",
923 threadId: projection.thread.id,
924 runId: run.id,
925 nodeId: rootNode.id,
926 providerInstanceId: run.providerInstanceId,
927 occurredAt: now,
928 payload: runningAttempt,
929 },
930 {
931 id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
932 type: "node.updated",
933 threadId: projection.thread.id,
934 runId: run.id,
935 nodeId: rootNode.id,
936 providerInstanceId: run.providerInstanceId,
937 occurredAt: now,
938 payload: runningRootNode,
939 },
940 ];
941 const runningWrite = yield* eventSink.writeIfRunCurrent({
942 threadId: projection.thread.id,
943 runId: run.id,
944 activeAttemptId: attempt.id,
945 expectedStatus: "starting",
946 events,
947 });
948 if (!runningWrite.committed) {
949 return;
950 }
951 const routableSubagents = projection.subagents.filter((subagent) =>
952 RunExecutionService.canRouteRelatedSubagent(subagent.status),
953 );
954 const userText = projectComposerContextForProvider({
955 text: message.text,
956 records: message.context?.records ?? [],
957 });
958 // Delivered once: this run's provider turn marks the work as told. A
959 // restart continuation is prompted by its own text or resumes natively.
960 const noteContinuation = isRestartNoteContinuation(
961 run,
962 projection.runs,
963 projection.providerTurns,
964 projection.attempts,
965 );
966 const restartCancelledWork = pendingRestartCancelledBackgroundWork({
967 runs: projection.runs,
968 providerTurns: projection.providerTurns,
969 compactionMessageIds: new Set(
970 projection.messages
971 .filter(
972 (candidate) =>
973 candidate.attachments.length === 0 &&
974 candidate.text.trim().toLowerCase() === "/compact",
975 )
976 .map((candidate) => candidate.id),
977 ),
978 run,
979 attempts: projection.attempts,
980 });
981 const restartNote =
982 restartCancelledWork.length === 0
983 ? ""
984 : restartCancelledBackgroundWorkNote(restartCancelledWork);
985 const tokenCap = yield* handoffTokenCapConfig.pipe(
986 Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP),
987 );
988 const settledHandoffs = projection.contextHandoffs.filter(
989 (handoff) =>
990 handoff.toProviderThreadId === providerThread.id &&
991 handoff.delivery?.nativeThreadId === runningProviderThread.nativeThreadRef?.nativeId &&
992 handoff.delivery?.status !== "pending",
993 );
994 const deliveredItemIds = new Set(
995 settledHandoffs.flatMap((handoff) => handoff.delivery?.itemIds ?? []),
996 );
997 const coveredItemIds = new Set([
998 ...deliveredItemIds,
999 ...settledHandoffs.flatMap((handoff) => handoff.delivery?.omittedItemIds ?? []),
1000 ]);
1001 const deliveredAttemptIds = new Set(
1002 projection.providerTurns.map((turn) => turn.runAttemptId),
1003 );
1004 const acceptedAttempts = projection.attempts.filter(
1005 (source) =>
1006 source.providerThreadId === providerThread.id && deliveredAttemptIds.has(source.id),
1007 );
1008 const nativeInputRunIds = new Set(
1009 acceptedAttempts
1010 .filter(
1011 (source) =>
1012 source.nativeThreadId !== undefined &&
1013 source.nativeThreadId === runningProviderThread.nativeThreadRef?.nativeId,
1014 )
1015 .map((source) => source.runId),
1016 );
1017 const legacyInputRunIds = new Set(
1018 acceptedAttempts
1019 .filter((source) => source.nativeThreadId === undefined)
1020 .map((source) => source.runId),
1021 );
1022 const legacyRecoveredRunIds = new Set(
1023 projection.runs
1024 .filter(
1025 (source) =>
1026 source.providerThreadId === providerThread.id &&
1027 settledHandoffs.some(
1028 (handoff) =>
1029 handoff.strategy === "full_thread_summary" &&
1030 handoff.fromProviderThreadIds.includes(providerThread.id) &&
1031 source.ordinal >= handoff.coveredRunOrdinals.from &&
1032 source.ordinal <= handoff.coveredRunOrdinals.to,
1033 ),
1034 )
1035 .map((source) => source.id),
1036 );
1037 // Use saved text and actual native attachments when telemetry is absent.
1038 // Legacy attempts lack native identity; exclude their explicitly recovered
1039 // history, whose attachments were not replayed into the replacement thread.
1040 const nativeContextEstimate = Effect.gen(function* () {
1041 return sameNativeThread
1042 ? (yield* projectionStore.getTurnStartHistory(input.threadId)).reduce((sum, item) => {
1043 if (
1044 item.runId === run.id ||
1045 (item.runId !== null &&
1046 missedRunIds.has(item.runId) &&
1047 !deliveredItemIds.has(item.id)) ||
1048 (item.providerThreadId !== providerThread.id && !deliveredItemIds.has(item.id))
1049 )
1050 return sum;
1051 const historical = historicalMessage(item);
1052 const nativeAttachments =
1053 item.type === "user_message" &&
1054 item.providerThreadId === providerThread.id &&
1055 item.runId !== null &&
1056 (nativeInputRunIds.has(item.runId) ||
1057 (legacyInputRunIds.has(item.runId) &&
1058 !coveredItemIds.has(item.id) &&
1059 !legacyRecoveredRunIds.has(item.runId)))
1060 ? attachmentTokenAllowance(item.attachments)
1061 : 0;
1062 return (
1063 sum +
1064 (historical === null ? 0 : Buffer.byteLength(historical.text)) +
1065 nativeAttachments
1066 );
1067 }, 0)
1068 : 0;
1069 });
1070 const modelContextWindow =
1071 knownModelWindow ??
1072 (handoffUsage !== null || reuseTelemetry ? previousUsage?.maxTokens : undefined);
1073 // Replacing a native thread clears its usage, not the selected model's capacity.
1074 const budgetProviderThread = {
1075 ...runningProviderThread,
1076 contextUsage: handoffUsage,
1077 };
1078 const missedRuns = projection.runs.filter(
1079 (source) =>
1080 source.ordinal < run.ordinal &&
1081 source.providerThreadId === providerThread.id &&
1082 (source.status === "failed" || source.status === "interrupted") &&
1083 !deliveredAttemptIds.has(source.activeAttemptId),
1084 );
1085 const missedRunIds = new Set(missedRuns.map((source) => source.id));
1086 const missedItems =
1087 missedRunIds.size === 0
1088 ? []
1089 : (yield* projectionStore.getTurnStartHistory(input.threadId, [...missedRunIds])).filter(
1090 (item) =>
1091 item.runId !== null &&
1092 missedRunIds.has(item.runId) &&
1093 !coveredItemIds.has(item.id) &&
1094 historicalMessage(item) !== null,
1095 );
1096 const startWithHandoffs = (
1097 turnInput: Parameters<typeof session.startTurn>[0],
1098 compact = false,
1099 ) =>
1100 Effect.gen(function* () {
1101 // A failed turn/start can leave the requested turn absent from
1102 // native history even when its preceding handoff was injected.
1103 const retryHandoff =
1104 missedItems.length === 0
1105 ? []
1106 : [
1107 yield* contextHandoffService.prepareProviderHandoff({
1108 threadId: projection.thread.id,
1109 targetRunId: run.id,
1110 transferId: null,
1111 fromProviderThreadIds: [providerThread.id],
1112 toProviderThreadId: providerThread.id,
1113 fromProviderInstanceId: run.providerInstanceId,
1114 toProviderInstanceId: run.providerInstanceId,
1115 coveredRunOrdinals: {
1116 from: missedRuns[0]!.ordinal,
1117 to: missedRuns.at(-1)!.ordinal,
1118 },
1119 strategy: "delta_since_target_last_seen",
1120 items: missedItems,
1121 runs: projection.runs,
1122 createdAt: yield* DateTime.now,
1123 }),
1124 ];
1125 const delivery = yield* deliverContextHandoffs({
1126 handoffs: [...effectiveHandoffs, ...retryHandoff],
1127 deferInline: compact,
1128 providerThread: runningProviderThread,
1129 budget: Effect.gen(function* () {
1130 return handoffBudget({
1131 tokenCap,
1132 modelContextWindow,
1133 // The note is sent with the user text, so it spends the same allowance.
1134 userText: restartNote === "" ? userText : `${restartNote}\n\n${userText}`,
1135 attachments: message.attachments,
1136 providerThread: budgetProviderThread,
1137 nativeContextEstimate:
1138 budgetProviderThread.contextUsage?.usedTokens === undefined
1139 ? yield* nativeContextEstimate
1140 : 0,
1141 });
1142 }),
1143 alreadyDeliveredItemIds: deliveredItemIds,
1144 ...(session.injectHistory === undefined
1145 ? {}
1146 : {
1147 inject: (history: ProviderAdapter.ProviderAdapterV2HistoricalContext) =>
1148 session.injectHistory!({
1149 providerThread: runningProviderThread,
1150 ...history,
1151 }),
1152 }),
1153 persist: (handoff) =>
1154 Effect.gen(function* () {
1155 const updatedAt = yield* DateTime.now;
1156 yield* eventSink.write({
1157 events: [
1158 {
1159 id: yield* idAllocator.allocate.event({
1160 threadId: projection.thread.id,
1161 }),
1162 type: "context-handoff.updated",
1163 threadId: projection.thread.id,
1164 runId: run.id,
1165 providerInstanceId: run.providerInstanceId,
1166 occurredAt: updatedAt,
1167 payload: { ...handoff, updatedAt },
1168 },
1169 ],
1170 });
1171 }),
1172 });
1173 if (!(yield* isCurrentAttemptInStatus("running"))) return;
1174 const start = compact ? session.compactThread! : session.startTurn;
1175 const context = [delivery.context, restartNote]
1176 .filter((part) => part !== "")
1177 .join("\n\n");
1178 // A note continuation has no turn to resume; its text is the prompt.
1179 const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput;
1180 yield* start({
1181 ...(noteContinuation ? promptedInput : turnInput),
1182 message: {
1183 ...turnInput.message,
1184 text: context === "" ? userText : `${context}\n\nUser message:\n${userText}`,
1185 },
1186 }).pipe(
1187 // A pending marker would make the next turn abandon this native
1188 // session, though the refused prompt never reached it.
1189 Effect.tapError((error) =>
1190 refusedBeforePrompt(error)
1191 ? delivery.unsent.pipe(
1192 Effect.catchCause(() =>
1193 Effect.logWarning("Failed to restore unsent context handoffs", {
1194 runId: run.id,
1195 }),
1196 ),
1197 )
1198 : Effect.void,
1199 ),
1200 );
1201 // The provider already accepted the turn. A stale pending marker
1202 // can force a fresh thread later, but must not stop live ingestion.
1203 yield* delivery.delivered.pipe(
1204 Effect.catchCause(() =>
1205 Effect.logWarning("Failed to record accepted context handoff delivery", {
1206 runId: run.id,
1207 deliveryStatus: "pending",
1208 }),
1209 ),
1210 );
1211 }).pipe(
1212 Effect.mapError((cause) =>
1213 cause._tag === "ProviderAdapterTurnStartError"
1214 ? cause
1215 : new ProviderAdapter.ProviderAdapterTurnStartError({
1216 driver: session.driver,
1217 threadId: projection.thread.id,
1218 providerThreadId: providerThread.id,
1219 runId: run.id,
1220 cause,
1221 }),
1222 ),
1223 );
1224 const deliverySession =
1225 effectiveHandoffs.length === 0 &&
1226 missedItems.length === 0 &&
1227 restartNote === "" &&
1228 !noteContinuation
1229 ? session
1230 : makeDeliverySession(session, startWithHandoffs);
1231 return runExecution.startRootRun({
1232 commandId: CommandId.make(`command:effect:provider-turn.start:${run.id}`),
1233 appThread: projection.thread,
1234 providerSessionId,
1235 session: deliverySession,
1236 run: runningRun,
1237 rootNode: runningRootNode,
1238 checkpointScope,
1239 providerThread: runningProviderThread,
1240 attempt: runningAttempt,
1241 attemptId: attempt.id,
1242 loadInheritedBackgroundTurnItems: runControls.loadInheritedBackgroundTurnItems,
1243 relatedThreadIds: routableSubagents.flatMap((subagent) =>
1244 subagent.childThreadId === null ? [] : [subagent.childThreadId],
1245 ),
1246 relatedProviderThreadIds: routableSubagents.flatMap((subagent) =>
1247 subagent.providerThreadId === null ? [] : [subagent.providerThreadId],
1248 ),
1249 providerTurnOrdinal:
1250 Math.max(
1251 0,
1252 ...projection.providerTurns
1253 .filter((turn) => turn.providerThreadId === providerThread.id)
1254 .map((turn) => turn.ordinal),
1255 ) + 1,
1256 // Legacy accepted attempts have no native id. They count only before
1257 // a replacement, while no accepted attempt records a native identity.
1258 nativeThreadHasTurns:
1259 nativeInputRunIds.size > 0 ||
1260 (legacyInputRunIds.size > 0 &&
1261 sameNativeThread &&
1262 !acceptedAttempts.some((source) => source.nativeThreadId !== undefined)),
1263 shouldStartProviderTurn: runControls.shouldStartProviderTurn,
1264 shouldFinalizeRun: runControls.shouldFinalizeRun,
1265 hasUnpairedRunInterruptRequest: runControls.hasUnpairedRunInterruptRequest,
1266 message: {
1267 messageId: message.id,
1268 text: userText,
1269 attachments: message.attachments,
1270 createdBy: message.createdBy,
1271 creationSource: message.creationSource,
1272 ...(message.scheduledTaskId === undefined
1273 ? {}
1274 : { scheduledTaskId: message.scheduledTaskId }),
1275 ...(message.senderThreadId === undefined
1276 ? {}
1277 : { senderThreadId: message.senderThreadId }),
1278 },
1279 modelSelection: run.modelSelection,
1280 runtimePolicy: resolvedRuntimePolicy,
1281 });
1282 });
1283
1284 return ProviderTurnStartServiceV2.of({
1285 start: (input) =>
1286 Effect.gen(function* () {
1287 // Storage cleanup removes worktrees under the same lease. Only startup
1288 // holds it; the provider turn runs after it is released.
1289 const { thread } = yield* projectionStore.getThreadRecords(input.threadId, []);
1290 const providerTurn = yield* thread.worktreePath === null
1291 ? start(input)
1292 : withWorkspaceLease(path.resolve(thread.worktreePath), start(input));
1293 if (providerTurn !== undefined) yield* providerTurn;
1294 }).pipe(
1295 Effect.mapError((cause) =>
1296 isProviderTurnStartError(cause)
1297 ? cause
1298 : new ProviderTurnStartError({ runId: input.runId, cause }),
1299 ),
1300 ),
1301 });
1302 }),
1303);