1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 EventId,
5 MessageId,
6 NodeId,
7 ProjectId,
8 ProviderDriverKind,
9 ProviderInstanceId,
10 ProviderThreadId,
11 ProviderTurnId,
12 RunAttemptId,
13 RunId,
14 ThreadId,
15 TurnItemId,
16} from "@t3tools/contracts";
17import * as DateTime from "effect/DateTime";
18import * as Effect from "effect/Effect";
19
20import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
21import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts";
22import * as EventSink from "./EventSink.ts";
23import * as Orchestrator from "./Orchestrator.ts";
24import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
25import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
26import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
27
28const forkCases = (["codex", "claudeAgent"] as const).flatMap((driverName) => {
29 const driver = ProviderDriverKind.make(driverName);
30 const instanceId = ProviderInstanceId.make(driver);
31 const modelSelection = { instanceId, model: "test-model" };
32 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
33 instanceId,
34 driver,
35 getCapabilities: () =>
36 Effect.succeed(
37 driver === "codex" ? CodexProviderCapabilitiesV2 : ClaudeProviderCapabilitiesV2,
38 ),
39 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
40 openSession: () => Effect.die("Execution is paused after dispatch for handoff inspection"),
41 };
42 const layer = ProviderReplayHarness.layerWithRegistry(
43 { name: `fork-boundary-${driver}` },
44 ProviderAdapterRegistry.layerFromAdapters([adapter]),
45 { runEffectWorker: false },
46 );
47
48 return (["failed", "interrupted", "cancelled"] as const).map((status) => ({
49 driver,
50 status,
51 instanceId,
52 modelSelection,
53 layer,
54 }));
55});
56
57it.effect.each(forkCases)(
58 "bounds $driver context when continuing a fork of a $status run",
59 ({ driver, status, instanceId, modelSelection, layer }) =>
60 Effect.gen(function* () {
61 const orchestrator = yield* Orchestrator.OrchestratorV2;
62 const eventSink = yield* EventSink.EventSinkV2;
63 const now = yield* DateTime.now;
64 const sourceThreadId = ThreadId.make("fork-boundary-source");
65 const targetThreadId = ThreadId.make("fork-boundary-target");
66 const providerThreadId = ProviderThreadId.make("fork-boundary-native-thread");
67 const sourceRunId = RunId.make("fork-boundary-source-run");
68 const attemptId = RunAttemptId.make("interrupted-source-attempt");
69 const providerTurnId = ProviderTurnId.make("interrupted-source-turn");
70 const rootNodeId = NodeId.make("interrupted-source-root");
71
72 yield* orchestrator.dispatch({
73 type: "thread.create",
74 commandId: CommandId.make("create-source"),
75 threadId: sourceThreadId,
76 projectId: ProjectId.make("fork-boundary-project"),
77 title: "Fork boundary source",
78 modelSelection,
79 runtimeMode: "full-access",
80 interactionMode: "default",
81 branch: null,
82 worktreePath: null,
83 createdBy: "user",
84 creationSource: "web",
85 });
86 yield* eventSink.write({
87 events: [
88 {
89 id: EventId.make("source-provider-thread"),
90 type: "provider-thread.updated",
91 threadId: sourceThreadId,
92 occurredAt: now,
93 payload: {
94 id: providerThreadId,
95 driver,
96 providerInstanceId: instanceId,
97 providerSessionId: null,
98 appThreadId: sourceThreadId,
99 ownerNodeId: null,
100 nativeThreadRef: { driver, nativeId: "native-source", strength: "strong" },
101 nativeConversationHeadRef: null,
102 status: "idle",
103 firstRunOrdinal: 1,
104 lastRunOrdinal: 2,
105 handoffIds: [],
106 forkedFrom: null,
107 createdAt: now,
108 updatedAt: now,
109 },
110 },
111 ],
112 });
113 // A cancelled queue entry has no provider turn; an early interruption
114 // can have a turn but no native assistant cursor.
115 if (status === "interrupted") {
116 yield* eventSink.write({
117 events: [
118 {
119 id: EventId.make("source-attempt"),
120 type: "run-attempt.created",
121 threadId: sourceThreadId,
122 runId: sourceRunId,
123 occurredAt: now,
124 payload: {
125 id: attemptId,
126 runId: sourceRunId,
127 attemptOrdinal: 1,
128 rootNodeId,
129 providerInstanceId: instanceId,
130 providerThreadId,
131 providerTurnId,
132 reason: "initial",
133 status,
134 startedAt: now,
135 completedAt: now,
136 },
137 },
138 {
139 id: EventId.make("source-provider-turn"),
140 type: "provider-turn.updated",
141 threadId: sourceThreadId,
142 occurredAt: now,
143 payload: {
144 id: providerTurnId,
145 providerThreadId,
146 nodeId: rootNodeId,
147 runAttemptId: attemptId,
148 nativeTurnRef: { driver, nativeId: "turn:synthetic", strength: "weak" },
149 ordinal: 1,
150 status,
151 startedAt: now,
152 completedAt: now,
153 },
154 },
155 ],
156 });
157 }
158 for (const ordinal of [1, 2]) {
159 const runId = ordinal === 1 ? sourceRunId : RunId.make("later-run");
160 const messageId = MessageId.make(`source-message-${ordinal}`);
161 yield* eventSink.write({
162 events: [
163 {
164 id: EventId.make(`run-${ordinal}`),
165 type: "run.created",
166 threadId: sourceThreadId,
167 runId,
168 occurredAt: now,
169 payload: {
170 id: runId,
171 threadId: sourceThreadId,
172 ordinal,
173 providerInstanceId: instanceId,
174 modelSelection,
175 providerThreadId,
176 userMessageId: messageId,
177 rootNodeId: null,
178 activeAttemptId: ordinal === 1 && status === "interrupted" ? attemptId : null,
179 status: ordinal === 1 ? status : "completed",
180 queuePosition: null,
181 requestedAt: now,
182 startedAt: now,
183 completedAt: now,
184 checkpointId: null,
185 contextHandoffId: null,
186 },
187 },
188 {
189 id: EventId.make(`item-${ordinal}`),
190 type: "turn-item.updated",
191 threadId: sourceThreadId,
192 runId,
193 occurredAt: now,
194 payload: {
195 id: TurnItemId.make(`item-${ordinal}`),
196 threadId: sourceThreadId,
197 runId,
198 nodeId: null,
199 providerThreadId,
200 providerTurnId: null,
201 nativeItemRef: null,
202 parentItemId: null,
203 ordinal,
204 status: "completed",
205 title: null,
206 startedAt: now,
207 completedAt: now,
208 updatedAt: now,
209 type: "user_message",
210 createdBy: "user",
211 creationSource: "web",
212 inputIntent: "turn_start",
213 messageId,
214 text: ordinal === 1 ? "INCLUDED_SOURCE_MARKER" : "EXCLUDED_LATER_MARKER",
215 attachments: [],
216 },
217 },
218 ],
219 });
220 }
221 yield* orchestrator.dispatch({
222 type: "thread.fork",
223 commandId: CommandId.make("fork-source"),
224 sourceThreadId,
225 targetThreadId,
226 sourcePoint: { type: "run", runId: sourceRunId },
227 createdBy: "user",
228 creationSource: "web",
229 });
230 yield* orchestrator.dispatch({
231 type: "message.dispatch",
232 commandId: CommandId.make("continue-fork"),
233 threadId: targetThreadId,
234 messageId: MessageId.make("continue-fork"),
235 text: "Continue from the selected source run",
236 attachments: [],
237 modelSelection,
238 dispatchMode: { type: "start_immediately" },
239 createdBy: "user",
240 creationSource: "web",
241 });
242 const target = yield* orchestrator.getThreadProjection(targetThreadId);
243 assert.equal(target.contextTransfers[0]?.resolution?.strategy, "portable_context");
244 assert.lengthOf(target.contextHandoffs, 1);
245 const handoff = target.contextHandoffs[0]!;
246 const history = handoff.history?.messages.map((message) => message.text).join("\n") ?? "";
247 assert.include(`${handoff.summaryText}\n${history}`, "INCLUDED_SOURCE_MARKER");
248 assert.notInclude(`${handoff.summaryText}\n${history}`, "EXCLUDED_LATER_MARKER");
249 assert.isNull(target.providerThreads[0]?.forkedFrom);
250 }).pipe(Effect.provide(layer)),
251);