1import {
2 ContextTransferId,
3 OrchestrationV2Actor,
4 OrchestrationV2AppThread,
5 OrchestrationV2ContextSourcePoint,
6 OrchestrationV2ContextTransfer,
7 OrchestrationV2CreationSource,
8 OrchestrationV2ProviderThread,
9 OrchestrationV2Run,
10 OrchestrationV2ThreadProjection,
11 ThreadId,
12} from "@t3tools/contracts";
13import * as Context from "effect/Context";
14import * as DateTime from "effect/DateTime";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Schema from "effect/Schema";
18
19export interface ThreadForkPlanV2 {
20 readonly targetThread: OrchestrationV2AppThread;
21 readonly transfer: OrchestrationV2ContextTransfer;
22}
23
24export class ThreadForkPlanError extends Schema.TaggedError<ThreadForkPlanError>()(
25 "ThreadForkPlanError",
26 {
27 sourceThreadId: ThreadId,
28 targetThreadId: ThreadId,
29 cause: Schema.optional(Schema.Defect()),
30 },
31) {}
32
33/**
34 * Fork copies a provider-finished conversation. Usage-limited and other
35 * failed, interrupted, or cancelled turns still have a native thread (or a
36 * portable transcript) even though the run did not complete successfully.
37 * `waiting` is provider-finished with checkpoint capture still pending.
38 * In-progress and rolled-back runs are not forkable.
39 */
40export function isForkableSourceRunStatus(status: OrchestrationV2Run["status"]): boolean {
41 return (
42 status === "completed" ||
43 status === "waiting" ||
44 status === "failed" ||
45 status === "interrupted" ||
46 status === "cancelled"
47 );
48}
49
50export function forkableSourceRunStatusError(
51 run: Pick<OrchestrationV2Run, "id" | "status">,
52): string {
53 return `Fork source run ${run.id} is ${run.status}; in-progress and rolled-back runs cannot be forked.`;
54}
55
56export interface ThreadForkServiceV2Shape {
57 readonly plan: (input: {
58 readonly sourceProjection: Pick<OrchestrationV2ThreadProjection, "thread">;
59 readonly sourceRun: OrchestrationV2Run;
60 readonly sourceProviderThread: OrchestrationV2ProviderThread | undefined;
61 readonly canonicalSourcePoint: OrchestrationV2ContextSourcePoint;
62 readonly transferId: ContextTransferId;
63 readonly targetThreadId: ThreadId;
64 readonly title?: string;
65 readonly createdBy: OrchestrationV2Actor;
66 readonly creationSource: OrchestrationV2CreationSource;
67 readonly createdAt: DateTime.Utc;
68 }) => Effect.Effect<ThreadForkPlanV2, ThreadForkPlanError>;
69}
70
71export class ThreadForkServiceV2 extends Context.Service<
72 ThreadForkServiceV2,
73 ThreadForkServiceV2Shape
74>()("t3/orchestration-v2/ThreadForkService/ThreadForkServiceV2") {}
75
76export const layer: Layer.Layer<ThreadForkServiceV2> = Layer.succeed(
77 ThreadForkServiceV2,
78 ThreadForkServiceV2.of({
79 plan: (input) =>
80 Effect.gen(function* () {
81 if (!isForkableSourceRunStatus(input.sourceRun.status)) {
82 return yield* new ThreadForkPlanError({
83 sourceThreadId: input.sourceProjection.thread.id,
84 targetThreadId: input.targetThreadId,
85 cause: forkableSourceRunStatusError(input.sourceRun),
86 });
87 }
88 const targetThread: OrchestrationV2AppThread = {
89 ...input.sourceProjection.thread,
90 createdBy: input.createdBy,
91 creationSource: input.creationSource,
92 id: input.targetThreadId,
93 title: input.title ?? `${input.sourceProjection.thread.title} fork`,
94 activeProviderThreadId: null,
95 lineage: {
96 parentThreadId: input.sourceProjection.thread.id,
97 relationshipToParent: "fork",
98 rootThreadId: input.sourceProjection.thread.lineage.rootThreadId,
99 },
100 forkedFrom: {
101 type: "run",
102 threadId: input.sourceProjection.thread.id,
103 runId: input.sourceRun.id,
104 },
105 createdAt: input.createdAt,
106 updatedAt: input.createdAt,
107 archivedAt: null,
108 settledOverride: null,
109 settledAt: null,
110 snoozedUntil: null,
111 snoozedAt: null,
112 lastVisitedAt: null,
113 deletedAt: null,
114 };
115 const transfer: OrchestrationV2ContextTransfer = {
116 id: input.transferId,
117 type: "fork",
118 sourceThreadId: input.sourceProjection.thread.id,
119 targetThreadId: input.targetThreadId,
120 sourcePoint: input.canonicalSourcePoint,
121 basePoint: null,
122 sourceProviderInstanceId: input.sourceRun.providerInstanceId,
123 targetProviderInstanceId: null,
124 targetRunId: null,
125 status: "pending",
126 resolution: null,
127 createdBy: input.createdBy,
128 error:
129 input.sourceProviderThread?.nativeThreadRef?.strength === "strong"
130 ? null
131 : "Source provider thread does not expose a strong native thread ref.",
132 createdAt: input.createdAt,
133 updatedAt: input.createdAt,
134 consumedAt: null,
135 };
136 return { targetThread, transfer };
137 }),
138 }),
139);