1import {
2 ContextTransferId,
3 OrchestrationV2Actor,
4 OrchestrationV2AppThread,
5 OrchestrationV2ContextSourcePoint,
6 OrchestrationV2ContextTransfer,
7 OrchestrationV2CreationSource,
8 OrchestrationV2ProviderThread,
9 OrchestrationV2ProviderTurnTokenUsage,
10 OrchestrationV2Run,
11 OrchestrationV2ThreadProjection,
12 ThreadId,
13} from "@t3tools/contracts";
14import * as Context from "effect/Context";
15import * as DateTime from "effect/DateTime";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18import * as Schema from "effect/Schema";
19
20export interface ThreadForkPlanV2 {
21 readonly targetThread: OrchestrationV2AppThread;
22 readonly transfer: OrchestrationV2ContextTransfer;
23}
24
25export class ThreadForkPlanError extends Schema.TaggedError<ThreadForkPlanError>()(
26 "ThreadForkPlanError",
27 {
28 sourceThreadId: ThreadId,
29 targetThreadId: ThreadId,
30 cause: Schema.optional(Schema.Defect()),
31 },
32) {}
33
34/** waiting means provider-finished with checkpoint capture pending. */
35export function isForkableSourceRunStatus(status: OrchestrationV2Run["status"]): boolean {
36 return (
37 status === "completed" ||
38 status === "waiting" ||
39 status === "failed" ||
40 status === "interrupted" ||
41 status === "cancelled"
42 );
43}
44
45export function forkableSourceRunStatusError(
46 run: Pick<OrchestrationV2Run, "id" | "status">,
47): string {
48 return `Fork source run ${run.id} is ${run.status}; in-progress and rolled-back runs cannot be forked.`;
49}
50
51export interface ThreadForkServiceV2Shape {
52 readonly plan: (input: {
53 readonly sourceProjection: Pick<OrchestrationV2ThreadProjection, "thread">;
54 /** Null only for a side chat whose parent has no stable run yet. */
55 readonly sourceRun: OrchestrationV2Run | null;
56 readonly sourceProviderThread: OrchestrationV2ProviderThread | undefined;
57 readonly sourceTokenUsage?: OrchestrationV2ProviderTurnTokenUsage | undefined;
58 readonly canonicalSourcePoint: OrchestrationV2ContextSourcePoint;
59 readonly transferId: ContextTransferId;
60 readonly targetThreadId: ThreadId;
61 readonly title?: string;
62 readonly sideChat?: boolean;
63 readonly createdBy: OrchestrationV2Actor;
64 readonly creationSource: OrchestrationV2CreationSource;
65 readonly createdAt: DateTime.Utc;
66 }) => Effect.Effect<ThreadForkPlanV2, ThreadForkPlanError>;
67}
68
69export class ThreadForkServiceV2 extends Context.Service<
70 ThreadForkServiceV2,
71 ThreadForkServiceV2Shape
72>()("t3/orchestration-v2/ThreadForkService/ThreadForkServiceV2") {}
73
74export const layer: Layer.Layer<ThreadForkServiceV2> = Layer.succeed(
75 ThreadForkServiceV2,
76 ThreadForkServiceV2.of({
77 plan: Effect.fn("ThreadForkService.plan")(function* (input) {
78 const { sourceRun } = input;
79 const sourceLineage = input.sourceProjection.thread.lineage;
80 // A task a linked environment delegated here is a subagent of a remote thread.
81 if (
82 input.sideChat &&
83 (sourceLineage.sideChat ||
84 sourceLineage.relationshipToParent === "subagent" ||
85 input.sourceProjection.thread.delegatedFrom !== undefined)
86 ) {
87 return yield* new ThreadForkPlanError({
88 sourceThreadId: input.sourceProjection.thread.id,
89 targetThreadId: input.targetThreadId,
90 cause: sourceLineage.sideChat
91 ? "A side chat cannot start another side chat."
92 : "A subagent cannot start a side chat.",
93 });
94 }
95 if (sourceRun !== null && !isForkableSourceRunStatus(sourceRun.status)) {
96 return yield* new ThreadForkPlanError({
97 sourceThreadId: input.sourceProjection.thread.id,
98 targetThreadId: input.targetThreadId,
99 cause: forkableSourceRunStatusError(sourceRun),
100 });
101 }
102 // The remote parent delegated the source thread, not its forks.
103 const { delegatedFrom: _delegatedFrom, ...inherited } = input.sourceProjection.thread;
104 const targetThread: OrchestrationV2AppThread = {
105 ...inherited,
106 // PR context follows the conversation; watching remains the source thread's responsibility.
107 pullRequests: input.sourceProjection.thread.pullRequests?.map(
108 ({ watch: _watch, ...link }) => link,
109 ),
110 createdBy: input.createdBy,
111 creationSource: input.creationSource,
112 id: input.targetThreadId,
113 title: input.title ?? `${input.sourceProjection.thread.title} fork`,
114 // The source's title job completes only on the source thread.
115 titleRegeneration: null,
116 activeProviderThreadId: null,
117 lineage: {
118 parentThreadId: input.sourceProjection.thread.id,
119 relationshipToParent: "fork",
120 rootThreadId: input.sourceProjection.thread.lineage.rootThreadId,
121 ...(input.sideChat ? { sideChat: true } : {}),
122 },
123 forkedFrom:
124 sourceRun === null
125 ? null
126 : { type: "run", threadId: input.sourceProjection.thread.id, runId: sourceRun.id },
127 createdAt: input.createdAt,
128 updatedAt: input.createdAt,
129 archivedAt: null,
130 settledOverride: null,
131 settledAt: null,
132 snoozedUntil: null,
133 snoozedAt: null,
134 lastSnoozeWakeAt: null,
135 pinnedAt: null,
136 pinOrderKey: null,
137 activeOrderKey: null,
138 autoSettleDisabledAt: null,
139 // A guest link belongs to the thread it was made for.
140 guestShare: null,
141 lastVisitedAt: null,
142 // A fork is a new thread; it does not inherit how often the source was opened.
143 opens: null,
144 deletedAt: null,
145 };
146 const transfer: OrchestrationV2ContextTransfer = {
147 id: input.transferId,
148 type: "fork",
149 sourceThreadId: input.sourceProjection.thread.id,
150 targetThreadId: input.targetThreadId,
151 sourcePoint: input.canonicalSourcePoint,
152 basePoint: null,
153 sourceProviderInstanceId:
154 sourceRun?.providerInstanceId ?? input.sourceProjection.thread.providerInstanceId,
155 // Only native forks retain this context; portable handoffs rebuild it.
156 sourceTokenUsage:
157 (sourceRun?.status === "completed" || sourceRun?.status === "waiting") &&
158 input.sourceProviderThread?.nativeThreadRef?.strength === "strong"
159 ? input.sourceTokenUsage
160 : undefined,
161 targetProviderInstanceId: null,
162 targetRunId: null,
163 status: "pending",
164 resolution: null,
165 createdBy: input.createdBy,
166 error:
167 input.sourceProviderThread?.nativeThreadRef?.strength === "strong"
168 ? null
169 : "Source provider thread does not expose a strong native thread ref.",
170 createdAt: input.createdAt,
171 updatedAt: input.createdAt,
172 consumedAt: null,
173 };
174 return { targetThread, transfer };
175 }),
176 }),
177);