1import type {
2 OrchestrationV2SearchThreadInput,
3 OrchestrationV2SearchThreadResult,
4} from "@t3tools/contracts";
5import type {
6 ProjectionRecordField,
7 ProjectionRecordFilter,
8 ProjectionRecords,
9} from "./ProjectionStore.ts";
10import {
11 type ChatAttachment,
12 CommandId,
13 MessageId,
14 OrchestrationV2StopWorkflowError,
15 type OrchestrationV2StopWorkflowInput,
16 type ModelSelection,
17 type OrchestrationV2Actor,
18 type OrchestrationV2Command,
19 type OrchestrationV2GetTurnItemResult,
20 type OrchestrationV2ServerCommand,
21 type OrchestrationV2ConversationMessage,
22 type OrchestrationV2CreationSource,
23 type OrchestrationV2Run,
24 type OrchestrationV2ThreadShellSnapshot,
25 type OrchestrationV2ThreadProjection,
26 type OrchestrationV2ThreadTranscript,
27 type OrchestrationV2ThreadShell,
28 type OrchestrationV2TurnItem,
29 ProjectId,
30 PROVIDER_SEND_TURN_MAX_FILE_BYTES,
31 ProviderInstanceId,
32 resolveProjectProviderInstanceEnabled,
33 RunId,
34 type ScheduledTaskId,
35 ThreadId,
36 type TurnItemId,
37} from "@t3tools/contracts";
38import { threadTranscriptHeader } from "@t3tools/shared/threadTranscript";
39import * as Context from "effect/Context";
40import * as DateTime from "effect/DateTime";
41import * as Duration from "effect/Duration";
42import * as Effect from "effect/Effect";
43import * as Layer from "effect/Layer";
44import * as Option from "effect/Option";
45import * as Ref from "effect/Ref";
46import * as Schema from "effect/Schema";
47import * as Stream from "effect/Stream";
48
49import * as CommandReceiptStore from "./CommandReceiptStore.ts";
50import * as Orchestrator from "./Orchestrator.ts";
51import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
52import { projectedRowEncodedBytes } from "./threadHistoryPaging.ts";
53import * as ProviderSessionManager from "./ProviderSessionManager.ts";
54import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
55import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimit.ts";
56import { projectTurnItemForDetail } from "./WireProjection.ts";
57import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
58import * as ServerSettings from "../serverSettings.ts";
59
60export type ThreadManagementSendMode = "auto" | "queue" | "steer" | "restart";
61
62export interface ThreadManagementProvenance {
63 readonly createdBy: OrchestrationV2Actor;
64 readonly creationSource: OrchestrationV2CreationSource;
65}
66
67export function withCreationProvenance(
68 command: OrchestrationV2Command,
69 provenance: ThreadManagementProvenance,
70): OrchestrationV2Command {
71 switch (command.type) {
72 case "thread.create":
73 case "message.dispatch":
74 case "thread.fork":
75 case "thread.merge_back":
76 case "delegated_task.request":
77 return { ...command, ...provenance };
78 default:
79 return command;
80 }
81}
82
83export function existingThreadIdsForCommand(
84 command: OrchestrationV2ServerCommand,
85): ReadonlyArray<ThreadId> {
86 switch (command.type) {
87 case "thread.create":
88 return [];
89 // Read-state commands only rewrite the thread payload's visited/unread
90 // watermark; they never touch messages, so they do not need the imported
91 // v1 transcript hydrated first. Visits fire on every thread-activity bump
92 // while a thread is open, so keeping them off the import path matters.
93 case "thread.visit":
94 case "thread.mark-unread":
95 return [];
96 case "thread.fork":
97 return [command.sourceThreadId];
98 case "thread.merge_back":
99 return command.sourceThreadId === command.targetThreadId
100 ? [command.sourceThreadId]
101 : [command.sourceThreadId, command.targetThreadId];
102 case "delegated_task.request":
103 case "delegated_task.wake-policy":
104 case "delegated_task.completion-delivery.acknowledge":
105 case "delegated_task.completion-delivery.dispose":
106 return [command.parentThreadId];
107 case "thread.created.record":
108 return command.parentThreadId === command.targetThreadId
109 ? [command.parentThreadId]
110 : [command.parentThreadId, command.targetThreadId];
111 default:
112 return [command.threadId];
113 }
114}
115
116export type ThreadManagementTerminalRunStatus = Extract<
117 OrchestrationV2Run["status"],
118 "completed" | "failed" | "cancelled" | "interrupted" | "rolled_back"
119>;
120
121export interface ThreadManagementSendInput {
122 readonly projectId: ProjectId;
123 readonly commandId: CommandId;
124 readonly threadId: ThreadId;
125 readonly messageId: MessageId;
126 readonly scheduledTaskId?: ScheduledTaskId;
127 readonly senderThreadId?: ThreadId;
128 readonly text: string;
129 readonly attachments: ReadonlyArray<ChatAttachment>;
130 readonly modelSelection?: ModelSelection;
131 readonly mode: ThreadManagementSendMode;
132 readonly createdBy: OrchestrationV2Actor;
133 readonly creationSource: OrchestrationV2CreationSource;
134}
135
136export interface ThreadManagementSendResult {
137 readonly dispatch: Orchestrator.OrchestratorV2DispatchResult;
138 readonly projection: Pick<
139 OrchestrationV2ThreadProjection,
140 "thread" | "runs" | "messages" | "turnItems"
141 >;
142 readonly message: OrchestrationV2ConversationMessage;
143 readonly run: OrchestrationV2Run;
144 /** Null for queued sends: the user turn item materializes when the queued turn starts. */
145 readonly turnItem: Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }> | null;
146 readonly delivery: "started" | "queued" | "steered" | "restarted";
147}
148
149export interface ThreadManagementWaitInput {
150 readonly projectId: ProjectId;
151 readonly threadId: ThreadId;
152 readonly runId?: RunId;
153 readonly timeoutMs: number;
154}
155
156export interface ThreadManagementWaitResult {
157 readonly threadId: ThreadId;
158 readonly run: OrchestrationV2Run | null;
159 readonly timedOut: boolean;
160}
161
162export interface ThreadManagementInterruptInput {
163 readonly projectId: ProjectId;
164 readonly commandId: CommandId;
165 readonly threadId: ThreadId;
166 readonly runId?: RunId;
167 readonly reason?: string;
168}
169
170export type ThreadManagementInterruptResult =
171 | {
172 readonly type: "interrupt_requested";
173 readonly run: OrchestrationV2Run;
174 readonly dispatch: Orchestrator.OrchestratorV2DispatchResult;
175 }
176 | { readonly type: "no_active_run" }
177 | {
178 readonly type: "already_terminal";
179 readonly run: OrchestrationV2Run & { readonly status: ThreadManagementTerminalRunStatus };
180 };
181
182export class ThreadManagementThreadNotFoundError extends Schema.TaggedError<ThreadManagementThreadNotFoundError>()(
183 "ThreadManagementThreadNotFoundError",
184 {
185 projectId: ProjectId,
186 threadId: ThreadId,
187 },
188) {
189 override get message(): string {
190 return `Thread ${this.threadId} was not found in project ${this.projectId}.`;
191 }
192}
193
194export class ThreadTranscriptTooLargeError extends Schema.TaggedError<ThreadTranscriptTooLargeError>()(
195 "ThreadTranscriptTooLargeError",
196 { threadId: ThreadId },
197) {
198 override get message(): string {
199 return "The thread transcript exceeds the file attachment limit.";
200 }
201}
202
203export class ThreadManagementRunNotFoundError extends Schema.TaggedError<ThreadManagementRunNotFoundError>()(
204 "ThreadManagementRunNotFoundError",
205 {
206 threadId: ThreadId,
207 runId: RunId,
208 },
209) {
210 override get message(): string {
211 return `Run ${this.runId} does not belong to thread ${this.threadId}.`;
212 }
213}
214
215export class ThreadManagementThreadArchivedError extends Schema.TaggedError<ThreadManagementThreadArchivedError>()(
216 "ThreadManagementThreadArchivedError",
217 {
218 threadId: ThreadId,
219 },
220) {
221 override get message(): string {
222 return `Thread ${this.threadId} is archived and cannot receive messages.`;
223 }
224}
225
226export class ThreadManagementNoSteerableRunError extends Schema.TaggedError<ThreadManagementNoSteerableRunError>()(
227 "ThreadManagementNoSteerableRunError",
228 {
229 threadId: ThreadId,
230 mode: Schema.Literals(["steer", "restart"]),
231 },
232) {
233 override get message(): string {
234 return `Thread ${this.threadId} has no running turn that can be ${this.mode === "steer" ? "steered" : "restarted"}.`;
235 }
236}
237
238export class ThreadManagementThreadNotInterruptibleError extends Schema.TaggedError<ThreadManagementThreadNotInterruptibleError>()(
239 "ThreadManagementThreadNotInterruptibleError",
240 {
241 threadId: ThreadId,
242 runId: RunId,
243 },
244) {
245 override get message(): string {
246 return `Run ${this.runId} is not currently interruptible.`;
247 }
248}
249
250export class ThreadManagementProjectionLoadError extends Schema.TaggedError<ThreadManagementProjectionLoadError>()(
251 "ThreadManagementProjectionLoadError",
252 {
253 projectId: ProjectId,
254 threadId: ThreadId,
255 cause: Schema.Defect(),
256 },
257) {
258 override get message(): string {
259 return `Unable to load thread ${this.threadId} in project ${this.projectId}.`;
260 }
261}
262
263export class ThreadManagementProjectThreadsListError extends Schema.TaggedError<ThreadManagementProjectThreadsListError>()(
264 "ThreadManagementProjectThreadsListError",
265 {
266 projectId: ProjectId,
267 cause: Schema.Defect(),
268 },
269) {
270 override get message(): string {
271 return `Unable to list threads in project ${this.projectId}.`;
272 }
273}
274
275export class ThreadManagementDurableRunProjectionError extends Schema.TaggedError<ThreadManagementDurableRunProjectionError>()(
276 "ThreadManagementDurableRunProjectionError",
277 {
278 threadId: ThreadId,
279 messageId: MessageId,
280 },
281) {
282 override get message(): string {
283 return `Message ${this.messageId} was accepted on thread ${this.threadId} without a durable run projection.`;
284 }
285}
286
287export class ThreadManagementProviderInstanceDisabledError extends Schema.TaggedError<ThreadManagementProviderInstanceDisabledError>()(
288 "ThreadManagementProviderInstanceDisabledError",
289 {
290 projectId: ProjectId,
291 instanceId: ProviderInstanceId,
292 },
293) {
294 override get message(): string {
295 return `Provider instance "${this.instanceId}" is disabled for project ${this.projectId}.`;
296 }
297}
298
299export const ThreadManagementError = Schema.Union([
300 ThreadManagementThreadNotFoundError,
301 ThreadManagementRunNotFoundError,
302 ThreadManagementThreadArchivedError,
303 ThreadManagementNoSteerableRunError,
304 ThreadManagementThreadNotInterruptibleError,
305 ThreadManagementProjectionLoadError,
306 ThreadManagementProjectThreadsListError,
307 ThreadManagementDurableRunProjectionError,
308 ThreadManagementProviderInstanceDisabledError,
309]);
310export type ThreadManagementError = typeof ThreadManagementError.Type;
311
312type ThreadManagementFailure = ThreadManagementError | Orchestrator.OrchestratorV2Error;
313
314export interface ThreadManagementServiceShape {
315 readonly searchThreadStream: (
316 input: OrchestrationV2SearchThreadInput,
317 ) => Stream.Stream<OrchestrationV2SearchThreadResult, Orchestrator.OrchestratorV2Error>;
318 readonly searchThread: (
319 input: OrchestrationV2SearchThreadInput,
320 ) => Effect.Effect<OrchestrationV2SearchThreadResult, Orchestrator.OrchestratorV2Error>;
321 readonly ensureLegacyTranscript: (
322 threadId: ThreadId,
323 ) => Effect.Effect<void, LegacyV1ThreadImporter.LegacyV1ThreadImportError>;
324 /**
325 * The single chokepoint for commands that can start or wake provider work:
326 * `message.dispatch` and `delegated_task.request` are refused here when the
327 * resolved instance is off for the project, except a command id that
328 * already has a receipt, which only replays its recorded result.
329 */
330 readonly dispatch: (
331 command: OrchestrationV2ServerCommand,
332 ) => Effect.Effect<
333 Orchestrator.OrchestratorV2DispatchResult,
334 Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
335 >;
336 readonly getThreadHistoryPage: Orchestrator.OrchestratorV2["Service"]["getThreadHistoryPage"];
337 readonly getTimelinePage: Orchestrator.OrchestratorV2["Service"]["getTimelinePage"];
338 readonly getMessageCount: Orchestrator.OrchestratorV2["Service"]["getMessageCount"];
339 /**
340 * One turn item with the input and output the thread stream withholds,
341 * bounded for the wire. Clients fetch it when a tool row is expanded.
342 */
343 readonly getTurnItem: (input: {
344 readonly threadId: ThreadId;
345 readonly itemId: TurnItemId;
346 }) => Effect.Effect<OrchestrationV2GetTurnItemResult, Orchestrator.OrchestratorV2Error>;
347 readonly getThreadRecords: Orchestrator.OrchestratorV2["Service"]["getThreadRecords"];
348 readonly getThreadProjection: (
349 threadId: ThreadId,
350 ) => Effect.Effect<OrchestrationV2ThreadProjection, Orchestrator.OrchestratorV2Error>;
351 readonly getCheckpointContext: Orchestrator.OrchestratorV2["Service"]["getCheckpointContext"];
352 readonly getThreadSnapshot: Orchestrator.OrchestratorV2["Service"]["getThreadSnapshot"];
353 readonly getThreadTranscript: (
354 threadId: ThreadId,
355 ) => Effect.Effect<
356 OrchestrationV2ThreadTranscript,
357 Orchestrator.OrchestratorV2Error | ThreadTranscriptTooLargeError,
358 ServerEnvironment.ServerEnvironmentIdentity
359 >;
360 readonly getThreadSnapshotWindow: Orchestrator.OrchestratorV2["Service"]["getThreadSnapshotWindow"];
361 readonly getProjectThreadRecords: <K extends ProjectionRecordField>(
362 input: { readonly projectId: ProjectId; readonly threadId: ThreadId },
363 fields: ReadonlyArray<K>,
364 filter?: ProjectionRecordFilter,
365 ) => Effect.Effect<
366 ProjectionRecords<K>,
367 ThreadManagementProjectionLoadError | ThreadManagementThreadNotFoundError
368 >;
369 readonly getProjectThread: (input: {
370 readonly projectId: ProjectId;
371 readonly threadId: ThreadId;
372 }) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadManagementError>;
373 readonly getShellSnapshot: (options?: {
374 readonly location?: "active" | "archive";
375 }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, Orchestrator.OrchestratorV2Error>;
376 readonly readShellSnapshot: (options?: {
377 readonly location?: "active" | "archive";
378 }) => Effect.Effect<
379 Effect.Effect<OrchestrationV2ThreadShellSnapshot, Orchestrator.OrchestratorV2Error>,
380 Orchestrator.OrchestratorV2Error
381 >;
382 readonly getThreadShell: Orchestrator.OrchestratorV2["Service"]["getThreadShell"];
383 readonly listProjectThreads: (input: {
384 readonly projectId: ProjectId;
385 readonly includeSubagents: boolean;
386 }) => Effect.Effect<ReadonlyArray<OrchestrationV2ThreadShell>, ThreadManagementError>;
387 readonly sendToThread: (
388 input: ThreadManagementSendInput,
389 ) => Effect.Effect<ThreadManagementSendResult, ThreadManagementFailure>;
390 /**
391 * Refuses a new turn on an instance the project has effectively disabled.
392 * A command that already has a receipt was let through before, so its
393 * retry replays instead of being refused under newer settings.
394 */
395 readonly assertProviderInstanceEnabled: (input: {
396 readonly projectId: ProjectId;
397 readonly commandId: CommandId;
398 readonly instanceId: ProviderInstanceId;
399 }) => Effect.Effect<void, ThreadManagementProviderInstanceDisabledError>;
400 /**
401 * Effective enablement for candidate selection, without the receipt-replay
402 * exemption `assertProviderInstanceEnabled` grants an accepted command.
403 */
404 readonly isProviderInstanceEnabledForProject: (input: {
405 readonly projectId: ProjectId;
406 readonly instanceId: ProviderInstanceId;
407 }) => Effect.Effect<boolean>;
408 /**
409 * Whether `commandId` already has a receipt. A caller resolving a target
410 * ahead of a dispatch under this same command id can use it to skip
411 * project-enablement filtering: the dispatch will only replay the recorded
412 * result, never start anything new, so the filter would just refuse a
413 * retry the project's settings no longer allow fresh.
414 */
415 readonly hasReceipt: (commandId: CommandId) => Effect.Effect<boolean>;
416 readonly waitForThread: (
417 input: ThreadManagementWaitInput,
418 ) => Effect.Effect<ThreadManagementWaitResult, ThreadManagementError>;
419 /**
420 * Waits for `runId` to end, then settles the thread if the run completed.
421 * Settling stops the provider session, so an agent that settles its own
422 * thread uses this to wait for its turn to end. A run that ends any other
423 * way leaves the thread active, as does anything `thread.settle` refuses
424 * then, such as a queued message.
425 */
426 readonly settleAfterRun: (input: {
427 readonly projectId: ProjectId;
428 readonly threadId: ThreadId;
429 readonly runId: RunId;
430 }) => Effect.Effect<void, ThreadManagementFailure>;
431 /**
432 * Settles a thread. When the thread's own agent asks, it is mid-turn, so
433 * the thread settles through `settleAfterRun` once that turn ends.
434 */
435 readonly settleThread: (input: {
436 readonly threadId: ThreadId;
437 readonly commandId: CommandId;
438 readonly byOwnAgent: boolean;
439 }) => Effect.Effect<
440 { readonly sequence: number } | { readonly settlesWhenTurnEnds: true },
441 // Dispatches thread.settle, a command the enablement chokepoint never
442 // gates; the type stays accurate anyway.
443 Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
444 >;
445 readonly interruptThread: (
446 input: ThreadManagementInterruptInput,
447 ) => Effect.Effect<ThreadManagementInterruptResult, ThreadManagementFailure>;
448 /**
449 * Asks the provider to stop a running workflow coordinator. The provider's
450 * own terminal notification then settles the coordinator and its members.
451 */
452 readonly stopWorkflow: (
453 input: OrchestrationV2StopWorkflowInput,
454 ) => Effect.Effect<void, OrchestrationV2StopWorkflowError>;
455 /**
456 * Sends `thread.stop` to every delegated task under a thread, depth first. Their command IDs
457 * derive from `commandId`, so a retry repeats nothing that already stopped. A task that
458 * cannot be stopped does not keep the others running: all are tried, then it fails.
459 */
460 readonly stopDelegatedTasks: (input: {
461 readonly threadId: ThreadId;
462 readonly commandId: CommandId;
463 readonly reason?: string | undefined;
464 }) => Effect.Effect<
465 void,
466 // Dispatches thread.stop, a command the enablement chokepoint never
467 // gates; the type stays accurate anyway.
468 Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
469 >;
470 readonly getThreadEventSequence: Orchestrator.OrchestratorV2["Service"]["getThreadEventSequence"];
471 readonly recoverDelegatedTask: Orchestrator.OrchestratorV2["Service"]["recoverDelegatedTask"];
472 readonly delegatedTaskResultPending: Orchestrator.OrchestratorV2["Service"]["delegatedTaskResultPending"];
473 readonly streamStoredEvents: Orchestrator.OrchestratorV2["Service"]["streamStoredEvents"];
474 readonly streamStoredEventsFrom: Orchestrator.OrchestratorV2["Service"]["streamStoredEventsFrom"];
475 readonly streamDomainEvents: Orchestrator.OrchestratorV2["Service"]["streamDomainEvents"];
476}
477
478export class ThreadManagementService extends Context.Service<
479 ThreadManagementService,
480 ThreadManagementServiceShape
481>()("t3/orchestration-v2/ThreadManagementService") {}
482
483export function isActiveRun(run: OrchestrationV2Run): boolean {
484 return (
485 run.status === "preparing" ||
486 run.status === "starting" ||
487 run.status === "running" ||
488 run.status === "waiting"
489 );
490}
491
492export function isTerminalRunStatus(
493 status: OrchestrationV2Run["status"],
494): status is ThreadManagementTerminalRunStatus {
495 return (
496 status === "completed" ||
497 status === "failed" ||
498 status === "cancelled" ||
499 status === "interrupted" ||
500 status === "rolled_back"
501 );
502}
503
504export function latestRun(
505 projection: Pick<OrchestrationV2ThreadProjection, "runs">,
506): OrchestrationV2Run | undefined {
507 return projection.runs.toSorted((left, right) => right.ordinal - left.ordinal)[0];
508}
509
510/** The newest run in `run`'s restart continuation chain, or `run` itself. */
511function latestRestartContinuation(
512 runs: ReadonlyArray<OrchestrationV2Run>,
513 run: OrchestrationV2Run,
514): OrchestrationV2Run {
515 let current = run;
516 for (let hops = 0; hops < runs.length; hops++) {
517 const next = runs.find((candidate) => candidate.restartContinuationOfRunId === current.id);
518 if (next === undefined) break;
519 current = next;
520 }
521 return current;
522}
523
524export function latestActiveRun(
525 projection: Pick<OrchestrationV2ThreadProjection, "runs">,
526): OrchestrationV2Run | undefined {
527 return projection.runs
528 .filter(isActiveRun)
529 .toSorted((left, right) => right.ordinal - left.ordinal)[0];
530}
531
532function latestSteerableRun(
533 projection: Pick<OrchestrationV2ThreadProjection, "runs" | "providerTurns">,
534): OrchestrationV2Run | undefined {
535 return projection.runs
536 .filter(
537 (run) =>
538 run.status === "running" &&
539 run.activeAttemptId !== null &&
540 projection.providerTurns.some(
541 (turn) => turn.runAttemptId === run.activeAttemptId && turn.status === "running",
542 ),
543 )
544 .toSorted((left, right) => right.ordinal - left.ordinal)[0];
545}
546
547const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000;
548
549const make = Effect.gen(function* () {
550 const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor;
551 const orchestrator = yield* Orchestrator.OrchestratorV2;
552 const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
553 const layerScope = yield* Effect.scope;
554 const legacyImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
555 const serverSettings = yield* ServerSettings.ServerSettingsService;
556 const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
557
558 const assertProviderInstanceEnabled: ThreadManagementServiceShape["assertProviderInstanceEnabled"] =
559 (input) =>
560 receipts.getByCommandId(input.commandId).pipe(
561 Effect.orElseSucceed(() => Option.none()),
562 Effect.flatMap((receipt) =>
563 Option.isSome(receipt)
564 ? Effect.void
565 : serverSettings.getSettings.pipe(
566 Effect.orDie,
567 Effect.flatMap((current) =>
568 resolveProjectProviderInstanceEnabled(current, input.projectId, input.instanceId)
569 ? Effect.void
570 : new ThreadManagementProviderInstanceDisabledError({
571 projectId: input.projectId,
572 instanceId: input.instanceId,
573 }),
574 ),
575 ),
576 ),
577 );
578
579 const isProviderInstanceEnabledForProject: ThreadManagementServiceShape["isProviderInstanceEnabledForProject"] =
580 (input) =>
581 serverSettings.getSettings.pipe(
582 Effect.orDie,
583 Effect.map((current) =>
584 resolveProjectProviderInstanceEnabled(current, input.projectId, input.instanceId),
585 ),
586 );
587
588 const hasReceipt: ThreadManagementServiceShape["hasReceipt"] = (commandId) =>
589 receipts.getByCommandId(commandId).pipe(
590 Effect.orElseSucceed(() => Option.none()),
591 Effect.map(Option.isSome),
592 );
593
594 const ensureLegacyTranscript = Effect.fn(
595 "orchestrationV2.threadManagement.ensureLegacyTranscript",
596 )(function* (threadId: ThreadId) {
597 yield* legacyImporter.ensureTranscript(threadId).pipe(
598 Effect.tapError((cause) =>
599 Effect.logWarning("Unable to hydrate migrated v1 thread transcript", {
600 threadId,
601 cause,
602 }),
603 ),
604 );
605 });
606
607 const ensureProjectionTranscript = (threadId: ThreadId) =>
608 ensureLegacyTranscript(threadId).pipe(
609 Effect.mapError(
610 (cause) =>
611 new Orchestrator.OrchestratorProjectionError({
612 threadId,
613 cause,
614 }),
615 ),
616 );
617
618 const ensureCommandTranscripts = Effect.fn(
619 "orchestrationV2.threadManagement.ensureCommandTranscripts",
620 )(function* (command: OrchestrationV2ServerCommand) {
621 yield* Effect.forEach(
622 existingThreadIdsForCommand(command),
623 (threadId) => ensureLegacyTranscript(threadId),
624 { discard: true },
625 ).pipe(
626 Effect.mapError(
627 (cause) =>
628 new Orchestrator.OrchestratorDispatchError({
629 commandId: command.commandId,
630 commandType: command.type,
631 cause,
632 }),
633 ),
634 );
635 });
636
637 const getThreadProjection: ThreadManagementServiceShape["getThreadProjection"] = (threadId) =>
638 ensureProjectionTranscript(threadId).pipe(
639 Effect.andThen(orchestrator.getThreadProjection(threadId)),
640 );
641
642 const getCheckpointContext: ThreadManagementServiceShape["getCheckpointContext"] = (threadId) =>
643 ensureProjectionTranscript(threadId).pipe(
644 Effect.andThen(orchestrator.getCheckpointContext(threadId)),
645 );
646
647 const getThreadSnapshot: ThreadManagementServiceShape["getThreadSnapshot"] = (threadId) =>
648 ensureProjectionTranscript(threadId).pipe(
649 Effect.andThen(orchestrator.getThreadSnapshot(threadId)),
650 );
651 const getThreadTranscript = Effect.fn("orchestrationV2.threadManagement.getThreadTranscript")(
652 function* (threadId: ThreadId) {
653 const { projection } = yield* getThreadSnapshot(threadId);
654 const environment = yield* ServerEnvironment.ServerEnvironmentIdentity;
655 const transcript = {
656 threadId: projection.thread.id,
657 title: projection.thread.title,
658 updatedAt: projection.updatedAt,
659 items: projection.visibleTurnItems,
660 };
661 let sizeBytes = Buffer.byteLength(
662 threadTranscriptHeader(yield* environment.getEnvironmentId, transcript),
663 "utf8",
664 );
665 if (sizeBytes > PROVIDER_SEND_TURN_MAX_FILE_BYTES) {
666 return yield* new ThreadTranscriptTooLargeError({ threadId });
667 }
668 for (const row of transcript.items) {
669 sizeBytes += projectedRowEncodedBytes(row) + 1;
670 if (sizeBytes > PROVIDER_SEND_TURN_MAX_FILE_BYTES) {
671 return yield* new ThreadTranscriptTooLargeError({ threadId });
672 }
673 }
674 return transcript;
675 },
676 );
677 const getThreadSnapshotWindow: ThreadManagementServiceShape["getThreadSnapshotWindow"] = (
678 threadId,
679 options,
680 ) =>
681 ensureProjectionTranscript(threadId).pipe(
682 Effect.andThen(orchestrator.getThreadSnapshotWindow(threadId, options)),
683 );
684
685 // Every path that can start or wake a provider turn funnels through this
686 // dispatch, so resolving and checking the target instance here (instead of
687 // at each caller) is the one place "off" is enforced.
688 const assertDispatchTargetEnabled = (command: OrchestrationV2ServerCommand) => {
689 switch (command.type) {
690 case "message.dispatch":
691 return orchestrator.getThreadRecords(command.threadId, []).pipe(
692 Effect.flatMap((projection) =>
693 assertProviderInstanceEnabled({
694 projectId: projection.thread.projectId,
695 commandId: command.commandId,
696 instanceId: (command.modelSelection ?? projection.thread.modelSelection).instanceId,
697 }),
698 ),
699 );
700 case "delegated_task.request":
701 return orchestrator.getThreadRecords(command.parentThreadId, []).pipe(
702 Effect.flatMap((parent) =>
703 assertProviderInstanceEnabled({
704 projectId: parent.thread.projectId,
705 commandId: command.commandId,
706 instanceId: command.modelSelection.instanceId,
707 }),
708 ),
709 );
710 default:
711 return Effect.void;
712 }
713 };
714
715 const dispatch: ThreadManagementServiceShape["dispatch"] = (command) =>
716 ensureCommandTranscripts(command).pipe(
717 Effect.andThen(assertDispatchTargetEnabled(command)),
718 Effect.andThen(orchestrator.dispatch(command)),
719 );
720
721 const getProjectThread: ThreadManagementServiceShape["getProjectThread"] = (input) =>
722 getThreadProjection(input.threadId).pipe(
723 Effect.mapError(
724 (cause) =>
725 new ThreadManagementProjectionLoadError({
726 projectId: input.projectId,
727 threadId: input.threadId,
728 cause,
729 }),
730 ),
731 Effect.flatMap((projection) =>
732 projection.thread.projectId === input.projectId && projection.thread.deletedAt === null
733 ? Effect.succeed(projection)
734 : Effect.fail(
735 new ThreadManagementThreadNotFoundError({
736 projectId: input.projectId,
737 threadId: input.threadId,
738 }),
739 ),
740 ),
741 );
742
743 const getProjectThreadRecords: ThreadManagementServiceShape["getProjectThreadRecords"] = (
744 input,
745 fields,
746 filter,
747 ) =>
748 ensureProjectionTranscript(input.threadId)
749 .pipe(Effect.andThen(orchestrator.getThreadRecords(input.threadId, fields, filter)))
750 .pipe(
751 Effect.mapError(
752 (cause) =>
753 new ThreadManagementProjectionLoadError({
754 projectId: input.projectId,
755 threadId: input.threadId,
756 cause,
757 }),
758 ),
759 Effect.flatMap((projection) =>
760 projection.thread.projectId === input.projectId && projection.thread.deletedAt === null
761 ? Effect.succeed(projection)
762 : Effect.fail(
763 new ThreadManagementThreadNotFoundError({
764 projectId: input.projectId,
765 threadId: input.threadId,
766 }),
767 ),
768 ),
769 );
770
771 const listProjectThreads: ThreadManagementServiceShape["listProjectThreads"] = (input) =>
772 orchestrator.getShellSnapshot({ projectId: input.projectId, location: "active" }).pipe(
773 Effect.mapError(
774 (cause) =>
775 new ThreadManagementProjectThreadsListError({
776 projectId: input.projectId,
777 cause,
778 }),
779 ),
780 Effect.map((snapshot) =>
781 snapshot.threads
782 .filter(
783 (thread) =>
784 input.includeSubagents || thread.lineage.relationshipToParent !== "subagent",
785 )
786 .toSorted(
787 (left, right) =>
788 DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt) ||
789 right.id.localeCompare(left.id),
790 ),
791 ),
792 );
793
794 const sendToThread: ThreadManagementServiceShape["sendToThread"] = (input) =>
795 Effect.gen(function* () {
796 const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]);
797 if (target.thread.archivedAt !== null) {
798 return yield* new ThreadManagementThreadArchivedError({
799 threadId: input.threadId,
800 });
801 }
802 const steerableRun = latestSteerableRun(target);
803 let dispatchMode: Extract<
804 OrchestrationV2Command,
805 { readonly type: "message.dispatch" }
806 >["dispatchMode"];
807 if (input.mode === "steer" || input.mode === "restart") {
808 if (steerableRun === undefined) {
809 return yield* new ThreadManagementNoSteerableRunError({
810 threadId: input.threadId,
811 mode: input.mode,
812 });
813 }
814 dispatchMode = {
815 type: input.mode === "steer" ? "steer_active" : "restart_active",
816 targetRunId: steerableRun.id,
817 };
818 } else if (input.mode === "auto" && steerableRun !== undefined) {
819 dispatchMode = { type: "steer_active", targetRunId: steerableRun.id };
820 } else {
821 dispatchMode = {
822 type: input.mode === "queue" ? "queue_after_active" : "start_immediately",
823 };
824 }
825
826 const dispatched = yield* dispatch({
827 type: "message.dispatch",
828 commandId: input.commandId,
829 threadId: input.threadId,
830 messageId: input.messageId,
831 ...(input.scheduledTaskId === undefined ? {} : { scheduledTaskId: input.scheduledTaskId }),
832 ...(input.senderThreadId === undefined ? {} : { senderThreadId: input.senderThreadId }),
833 text: input.text,
834 attachments: input.attachments,
835 ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }),
836 dispatchMode,
837 createdBy: input.createdBy,
838 creationSource: input.creationSource,
839 });
840 const projection = yield* getProjectThreadRecords(input, ["runs", "messages", "turnItems"], {
841 messageIds: [input.messageId],
842 turnItemTypes: ["user_message"],
843 });
844 const message = projection.messages.find((candidate) => candidate.id === input.messageId);
845 const run =
846 message?.runId === null || message?.runId === undefined
847 ? undefined
848 : projection.runs.find((candidate) => candidate.id === message.runId);
849 const turnItem =
850 projection.turnItems.find(
851 (
852 candidate,
853 ): candidate is Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }> =>
854 candidate.type === "user_message" && candidate.messageId === input.messageId,
855 ) ?? null;
856 // A queued message's user turn item is deliberately not emitted at
857 // dispatch time — it materializes when the queued turn actually starts,
858 // so it can map onto the provider turn. Every other dispatch mode still
859 // produces its turn item transactionally with the run.
860 if (
861 message === undefined ||
862 run === undefined ||
863 (turnItem === null && run.status !== "queued")
864 ) {
865 return yield* new ThreadManagementDurableRunProjectionError({
866 threadId: input.threadId,
867 messageId: input.messageId,
868 });
869 }
870 const delivery: ThreadManagementSendResult["delivery"] =
871 turnItem === null || turnItem.inputIntent === "queued_turn"
872 ? "queued"
873 : turnItem.inputIntent === "turn_start"
874 ? "started"
875 : input.mode === "restart"
876 ? "restarted"
877 : "steered";
878 return { dispatch: dispatched, projection, message, run, turnItem, delivery };
879 });
880
881 const waitForThread: ThreadManagementServiceShape["waitForThread"] = (input) =>
882 Effect.gen(function* () {
883 const loadError = (cause: unknown) =>
884 new ThreadManagementProjectionLoadError({
885 projectId: input.projectId,
886 threadId: input.threadId,
887 cause,
888 });
889 // Taken before the first read, so a run update between that read and
890 // the subscription below still replays.
891 const afterSequence = yield* orchestrator
892 .getThreadEventSequence(input.threadId)
893 .pipe(Effect.mapError(loadError));
894 const target = yield* getProjectThreadRecords(input, ["runs"]);
895 const selectedRun =
896 input.runId === undefined
897 ? latestRun(target)
898 : target.runs.find((candidate) => candidate.id === input.runId);
899 if (input.runId !== undefined && selectedRun === undefined) {
900 return yield* new ThreadManagementRunNotFoundError({
901 threadId: input.threadId,
902 runId: input.runId,
903 });
904 }
905 if (selectedRun === undefined) {
906 return { threadId: input.threadId, run: null, timedOut: false };
907 }
908 // A restart cuts a run and continues it as a new run. Follow that chain,
909 // so a cut run is not the answer while its continuation is pending.
910 const outcome = (runs: ReadonlyArray<OrchestrationV2Run>) =>
911 Effect.gen(function* () {
912 const current = runs.find((candidate) => candidate.id === selectedRun.id);
913 if (current === undefined) {
914 return yield* new ThreadManagementRunNotFoundError({
915 threadId: input.threadId,
916 runId: selectedRun.id,
917 });
918 }
919 const run = latestRestartContinuation(runs, current);
920 const pending =
921 isTerminalRunStatus(run.status) &&
922 (yield* orchestrator.restartContinuationPending(run));
923 return { run, pending, settled: isTerminalRunStatus(run.status) && !pending };
924 });
925 const initial = yield* outcome(target.runs);
926 if (initial.settled) {
927 return { threadId: input.threadId, run: initial.run, timedOut: false };
928 }
929
930 // Re-read the runs only when the thread records an update to the waited
931 // run or a continuation, or its deletion, instead of polling the
932 // projection for up to an hour. Each stream keeps one event type, so
933 // transcript events never fill its buffer. A declined continuation
934 // records no event, so its effect settling wakes the wait too.
935 const wait = Stream.merge(
936 Stream.merge(
937 orchestrator.streamStoredEventsFrom({
938 threadId: input.threadId,
939 afterSequence,
940 eventType: "run.updated",
941 }),
942 orchestrator.streamStoredEventsFrom({
943 threadId: input.threadId,
944 afterSequence,
945 eventType: "thread.deleted",
946 }),
947 ).pipe(
948 Stream.mapError(loadError),
949 Stream.filter(
950 (stored) =>
951 stored.event.type !== "run.updated" ||
952 stored.event.payload.id === selectedRun.id ||
953 stored.event.payload.restartContinuationOfRunId !== undefined,
954 ),
955 ),
956 initial.pending
957 ? Stream.fromEffect(orchestrator.awaitRestartContinuation(initial.run))
958 : Stream.empty,
959 ).pipe(
960 Stream.mapEffect(() =>
961 getProjectThreadRecords(input, ["runs"]).pipe(
962 Effect.flatMap((current) => outcome(current.runs)),
963 ),
964 ),
965 Stream.filter((current) => current.settled),
966 Stream.runHead,
967 Effect.timeoutOption(Duration.millis(Math.max(1, input.timeoutMs))),
968 );
969 const waited = Option.flatten(yield* wait);
970 if (Option.isSome(waited)) {
971 return { threadId: input.threadId, run: waited.value.run, timedOut: false };
972 }
973 // The run may have reached a terminal status while the timeout was
974 // winning the race; the final projection read decides what actually
975 // happened, so only a still-active run counts as timed out.
976 const final = yield* outcome((yield* getProjectThreadRecords(input, ["runs"])).runs);
977 return { threadId: input.threadId, run: final.run, timedOut: !final.settled };
978 });
979
980 const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn(
981 "orchestrationV2.threadManagement.settleAfterRun",
982 )(function* (input) {
983 // A turn that outlives this wait leaves its thread active.
984 const { run } = yield* waitForThread({ ...input, timeoutMs: SETTLE_AFTER_RUN_WAIT_MS });
985 if (run?.status !== "completed") return;
986 yield* dispatch({
987 type: "thread.settle",
988 commandId: CommandId.make(`server:settle-after-run:${input.runId}`),
989 threadId: input.threadId,
990 });
991 });
992
993 const settleThread: ThreadManagementServiceShape["settleThread"] = Effect.fn(
994 "orchestrationV2.threadManagement.settleThread",
995 )(function* (input) {
996 const shell = input.byOwnAgent ? yield* orchestrator.getThreadShell(input.threadId) : null;
997 if (shell != null && shell.activeRunId !== null) {
998 yield* settleAfterRun({
999 projectId: shell.projectId,
1000 threadId: shell.id,
1001 runId: shell.activeRunId,
1002 }).pipe(Effect.ignoreCause({ log: true }), Effect.forkIn(layerScope));
1003 return { settlesWhenTurnEnds: true } as const;
1004 }
1005 const result = yield* dispatch({
1006 type: "thread.settle",
1007 commandId: input.commandId,
1008 threadId: input.threadId,
1009 });
1010 return { sequence: result.sequence };
1011 });
1012
1013 const interruptThread: ThreadManagementServiceShape["interruptThread"] = (input) =>
1014 Effect.gen(function* () {
1015 const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]);
1016 const explicitRun =
1017 input.runId === undefined
1018 ? undefined
1019 : target.runs.find((candidate) => candidate.id === input.runId);
1020 if (input.runId !== undefined && explicitRun === undefined) {
1021 return yield* new ThreadManagementRunNotFoundError({
1022 threadId: input.threadId,
1023 runId: input.runId,
1024 });
1025 }
1026 if (explicitRun !== undefined && isTerminalRunStatus(explicitRun.status)) {
1027 return {
1028 type: "already_terminal",
1029 run: explicitRun as OrchestrationV2Run & {
1030 readonly status: ThreadManagementTerminalRunStatus;
1031 },
1032 } as const;
1033 }
1034 const interruptibleRun = latestActiveRun(target);
1035 if (interruptibleRun === undefined) {
1036 if (input.runId === undefined) {
1037 return { type: "no_active_run" } as const;
1038 }
1039 return yield* new ThreadManagementThreadNotInterruptibleError({
1040 threadId: input.threadId,
1041 runId: input.runId,
1042 });
1043 }
1044 if (input.runId !== undefined && interruptibleRun.id !== input.runId) {
1045 return yield* new ThreadManagementThreadNotInterruptibleError({
1046 threadId: input.threadId,
1047 runId: input.runId,
1048 });
1049 }
1050 const dispatch = yield* orchestrator.dispatch({
1051 type: "run.interrupt",
1052 commandId: input.commandId,
1053 threadId: input.threadId,
1054 runId: interruptibleRun.id,
1055 ...(input.reason === undefined ? {} : { reason: input.reason }),
1056 });
1057 return { type: "interrupt_requested", run: interruptibleRun, dispatch } as const;
1058 });
1059
1060 const stopWorkflow: ThreadManagementServiceShape["stopWorkflow"] = (input) =>
1061 Effect.gen(function* () {
1062 const records = yield* orchestrator
1063 .getThreadRecords(input.threadId, ["subagents", "runs", "providerThreads"])
1064 .pipe(
1065 Effect.mapError(
1066 (cause) =>
1067 new OrchestrationV2StopWorkflowError({ ...input, reason: "unavailable", cause }),
1068 ),
1069 );
1070 if (records.thread.deletedAt !== null) {
1071 return yield* new OrchestrationV2StopWorkflowError({ ...input, reason: "unavailable" });
1072 }
1073 // A limited caller (an agent over MCP) may not stop work on a thread
1074 // running above its own modes. Checked under the thread lock, where the
1075 // thread's user cannot raise the modes between this read and the stop.
1076 const limit = yield* DispatchModeLimit;
1077 const exceeded =
1078 limit === undefined ? undefined : exceededDispatchModeLimit(limit, records.thread);
1079 if (limit !== undefined && exceeded !== undefined) {
1080 if (limit.refused !== undefined) {
1081 yield* Ref.set(limit.refused, {
1082 threadId: input.threadId,
1083 runtimeMode: records.thread.runtimeMode,
1084 interactionMode: records.thread.interactionMode,
1085 mode: exceeded,
1086 });
1087 }
1088 return yield* new OrchestrationV2StopWorkflowError({ ...input, reason: "unavailable" });
1089 }
1090 const task = records.subagents.find((candidate) => candidate.id === input.subagentId);
1091 if (
1092 task === undefined ||
1093 task.threadId !== input.threadId ||
1094 task.workflow === undefined ||
1095 task.status !== "running"
1096 ) {
1097 return yield* new OrchestrationV2StopWorkflowError({ ...input, reason: "not-running" });
1098 }
1099 if (task.origin !== "provider_native" || task.driver !== "claudeAgent") {
1100 return yield* new OrchestrationV2StopWorkflowError({ ...input, reason: "unsupported" });
1101 }
1102 const providerThreadId =
1103 task.providerThreadId ??
1104 records.runs.find((run) => run.id === task.runId)?.providerThreadId;
1105 const providerThread = records.providerThreads.find(
1106 (candidate) =>
1107 candidate.id === providerThreadId &&
1108 candidate.appThreadId === input.threadId &&
1109 candidate.providerInstanceId === task.providerInstanceId &&
1110 candidate.driver === task.driver,
1111 );
1112 const taskId = task.nativeTaskRef?.nativeId;
1113 if (providerThread?.providerSessionId == null || taskId == null || taskId.length === 0) {
1114 return yield* new OrchestrationV2StopWorkflowError({ ...input, reason: "unavailable" });
1115 }
1116 const runtime = yield* sessions
1117 .get(providerThread.providerSessionId)
1118 .pipe(
1119 Effect.mapError(
1120 (cause) =>
1121 new OrchestrationV2StopWorkflowError({ ...input, reason: "unavailable", cause }),
1122 ),
1123 );
1124 if (Option.isNone(runtime)) {
1125 return yield* new OrchestrationV2StopWorkflowError({ ...input, reason: "unavailable" });
1126 }
1127 if (runtime.value.stopTask === undefined) {
1128 return yield* new OrchestrationV2StopWorkflowError({ ...input, reason: "unsupported" });
1129 }
1130 // Bounded so a provider that never answers cannot hold the thread lock.
1131 yield* runtime.value.stopTask({ providerThread, taskId }).pipe(
1132 Effect.timeout("15 seconds"),
1133 Effect.mapError(
1134 (cause) =>
1135 new OrchestrationV2StopWorkflowError({
1136 ...input,
1137 reason: "stop-failed",
1138 cause,
1139 }),
1140 ),
1141 );
1142 }).pipe((effect) => threadDispatch.withLock(input.threadId, effect));
1143
1144 const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = (input) =>
1145 Effect.gen(function* () {
1146 const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]);
1147 const failures: Array<
1148 Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
1149 > = [];
1150 for (const task of subagents) {
1151 if (task.origin !== "app_owned" || task.childThreadId === null) continue;
1152 const threadId = task.childThreadId;
1153 yield* dispatch({
1154 type: "thread.stop",
1155 commandId: CommandId.make(`${input.commandId}:stop:${threadId}`),
1156 threadId,
1157 ...(input.reason === undefined ? {} : { reason: input.reason }),
1158 }).pipe(
1159 Effect.andThen(stopDelegatedTasks({ ...input, threadId })),
1160 Effect.catch((error) =>
1161 Effect.logWarning("Unable to stop a delegated task", {
1162 parentThreadId: input.threadId,
1163 threadId,
1164 error,
1165 }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))),
1166 ),
1167 );
1168 }
1169 if (failures[0] !== undefined) return yield* Effect.fail(failures[0]);
1170 });
1171
1172 return ThreadManagementService.of({
1173 ensureLegacyTranscript,
1174 searchThreadStream: (input) =>
1175 Stream.unwrap(
1176 ensureProjectionTranscript(input.threadId).pipe(
1177 Effect.as(orchestrator.searchThreadStream(input)),
1178 ),
1179 ),
1180 searchThread: (input) =>
1181 ensureProjectionTranscript(input.threadId).pipe(
1182 Effect.andThen(orchestrator.searchThread(input)),
1183 ),
1184 dispatch,
1185 getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) =>
1186 ensureProjectionTranscript(threadId).pipe(
1187 Effect.andThen(
1188 orchestrator.getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly),
1189 ),
1190 ),
1191 getTimelinePage: (threadId, options) =>
1192 ensureProjectionTranscript(threadId).pipe(
1193 Effect.andThen(orchestrator.getTimelinePage(threadId, options)),
1194 ),
1195 getMessageCount: (threadId) =>
1196 ensureProjectionTranscript(threadId).pipe(
1197 Effect.andThen(orchestrator.getMessageCount(threadId)),
1198 ),
1199 getTurnItem: (input) =>
1200 ensureProjectionTranscript(input.threadId).pipe(
1201 Effect.andThen(orchestrator.getTurnItem(input)),
1202 Effect.map((item) => ({ item: item === null ? null : projectTurnItemForDetail(item) })),
1203 ),
1204 getThreadRecords: (threadId, fields, filter) =>
1205 ensureProjectionTranscript(threadId).pipe(
1206 Effect.andThen(orchestrator.getThreadRecords(threadId, fields, filter)),
1207 ),
1208 getThreadProjection,
1209 getCheckpointContext,
1210 getThreadSnapshot,
1211 getThreadTranscript,
1212 getThreadSnapshotWindow,
1213 getProjectThreadRecords,
1214 getProjectThread,
1215 getShellSnapshot: orchestrator.getShellSnapshot,
1216 readShellSnapshot: orchestrator.readShellSnapshot,
1217 getThreadShell: orchestrator.getThreadShell,
1218 listProjectThreads,
1219 sendToThread,
1220 assertProviderInstanceEnabled,
1221 isProviderInstanceEnabledForProject,
1222 hasReceipt,
1223 waitForThread,
1224 settleAfterRun,
1225 settleThread,
1226 interruptThread,
1227 stopDelegatedTasks,
1228 stopWorkflow,
1229 getThreadEventSequence: orchestrator.getThreadEventSequence,
1230 recoverDelegatedTask: orchestrator.recoverDelegatedTask,
1231 delegatedTaskResultPending: orchestrator.delegatedTaskResultPending,
1232 streamStoredEvents: orchestrator.streamStoredEvents,
1233 streamStoredEventsFrom: orchestrator.streamStoredEventsFrom,
1234 streamDomainEvents: orchestrator.streamDomainEvents,
1235 });
1236});
1237
1238const layerLegacyV1ThreadImporterNoop = Layer.succeed(
1239 LegacyV1ThreadImporter.LegacyV1ThreadImporter,
1240 LegacyV1ThreadImporter.LegacyV1ThreadImporter.of({
1241 pendingThreadCount: Effect.succeed(0),
1242 reconcileShells: Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }),
1243 ensureTranscript: () => Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }),
1244 importPendingTranscripts: Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }),
1245 }),
1246);
1247
1248export const layer: Layer.Layer<
1249 ThreadManagementService,
1250 never,
1251 | Orchestrator.OrchestratorV2
1252 | ProviderSessionManager.ProviderSessionManagerV2
1253 | ServerSettings.ServerSettingsService
1254 | CommandReceiptStore.CommandReceiptStoreV2
1255> = Layer.effect(ThreadManagementService, make).pipe(
1256 Layer.provide(layerLegacyV1ThreadImporterNoop),
1257 Layer.provide(ThreadCommandExecutor.layer),
1258);
1259
1260export const layerWithLegacyImporter: Layer.Layer<
1261 ThreadManagementService,
1262 never,
1263 | LegacyV1ThreadImporter.LegacyV1ThreadImporter
1264 | Orchestrator.OrchestratorV2
1265 | ProviderSessionManager.ProviderSessionManagerV2
1266 | ServerSettings.ServerSettingsService
1267 | CommandReceiptStore.CommandReceiptStoreV2
1268> = Layer.effect(ThreadManagementService, make).pipe(Layer.provide(ThreadCommandExecutor.layer));