1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import {
4 CommandId,
5 MessageId,
6 type OrchestrationV2Command,
7 type OrchestrationV2ThreadProjection,
8 ProviderInstanceId,
9 type ProviderReplayTranscript,
10 ThreadId,
11} from "@t3tools/contracts";
12import * as Effect from "effect/Effect";
13import * as FileSystem from "effect/FileSystem";
14import * as Path from "effect/Path";
15import * as PlatformError from "effect/PlatformError";
16import * as Schema from "effect/Schema";
17import { ChildProcess, ChildProcessSpawner } from "effect/process";
18
19import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts";
20import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts";
21import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
22import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
23import {
24 THREAD_FORK_NATIVE_PRIOR_TURN_ALPHA_PROMPT,
25 THREAD_FORK_NATIVE_PRIOR_TURN_BETA_PROMPT,
26 THREAD_FORK_NATIVE_PRIOR_TURN_REPEAT_PROMPT,
27 THREAD_FORK_NATIVE_SOURCE_PROMPT,
28 THREAD_FORK_NATIVE_TARGET_PROMPT,
29} from "./fixtures/shared.ts";
30import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts";
31import {
32 decodeProviderReplayNdjson,
33 materializeReplayTranscriptWorkspace,
34} from "@t3tools/provider-testing/replayTranscript";
35
36const CODEX_MODEL_SELECTION = {
37 instanceId: ProviderInstanceId.make("codex"),
38 model: "gpt-5.4",
39} as const;
40const CLAUDE_MODEL_SELECTION = {
41 instanceId: ProviderInstanceId.make("claudeAgent"),
42 model: "claude-sonnet-4-6",
43} as const;
44const PRIOR_TURN_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_prior_turn/codex_transcript.ndjson`;
45const CLAUDE_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native/claude_transcript.ndjson`;
46const CLAUDE_PRIOR_TURN_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_prior_turn/claude_transcript.ndjson`;
47const CLAUDE_FORK_LOCAL_ROLLBACK_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_fork_local_rollback/claude_transcript.ndjson`;
48const CODEX_READ_ONLY_NEVER_POLICY = {
49 approvalPolicy: "never",
50 sandboxPolicy: {
51 type: "readOnly",
52 access: { type: "fullAccess" },
53 networkAccess: false,
54 },
55} as const;
56
57class ThreadForkGitCommandError extends Schema.TaggedError<ThreadForkGitCommandError>()(
58 "ThreadForkGitCommandError",
59 {
60 command: Schema.String,
61 exitCode: Schema.Number,
62 },
63) {
64 override get message(): string {
65 return `${this.command} failed with exit ${this.exitCode}.`;
66 }
67}
68
69function runGit(
70 cwd: string,
71 args: ReadonlyArray<string>,
72): Effect.Effect<
73 void,
74 ThreadForkGitCommandError | PlatformError.PlatformError,
75 ChildProcessSpawner.ChildProcessSpawner
76> {
77 return Effect.gen(function* () {
78 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
79 const exitCode = yield* spawner.exitCode(ChildProcess.make("git", args, { cwd }));
80 if (Number(exitCode) !== 0) {
81 return yield* new ThreadForkGitCommandError({
82 command: `git ${args.join(" ")}`,
83 exitCode: Number(exitCode),
84 });
85 }
86 });
87}
88
89const makeCheckpointWorkspace = Effect.gen(function* () {
90 const fs = yield* FileSystem.FileSystem;
91 const path = yield* Path.Path;
92 const cwd = yield* fs.makeTempDirectory({ prefix: "t3-orchestrator-v2-thread-fork-" });
93 yield* runGit(cwd, ["init"]);
94 yield* runGit(cwd, ["config", "user.name", "T3 Code Test"]);
95 yield* runGit(cwd, ["config", "user.email", "t3code-test@example.com"]);
96 yield* fs.writeFileString(path.join(cwd, "README.md"), "# thread fork\n");
97 yield* runGit(cwd, ["add", "README.md"]);
98 yield* runGit(cwd, ["commit", "-m", "initial"]);
99 return cwd;
100});
101
102function readTranscript(transcriptPath: string) {
103 return Effect.gen(function* () {
104 const fs = yield* FileSystem.FileSystem;
105 const text = yield* fs.readFileString(transcriptPath);
106 return yield* decodeProviderReplayNdjson(text);
107 });
108}
109
110function metadataString(transcript: ProviderReplayTranscript, key: string): string {
111 const value = transcript.metadata?.[key];
112 if (typeof value !== "string") {
113 throw new Error(`Transcript ${transcript.scenario} is missing metadata string ${key}.`);
114 }
115 return value;
116}
117
118function metadataStringArray(
119 transcript: ProviderReplayTranscript,
120 key: string,
121): ReadonlyArray<string> {
122 const value = transcript.metadata?.[key];
123 if (!Array.isArray(value) || !value.every((entry) => typeof entry === "string")) {
124 throw new Error(`Transcript ${transcript.scenario} is missing metadata string array ${key}.`);
125 }
126 return value;
127}
128
129function userAndAssistantText(
130 projection: Pick<OrchestrationV2ThreadProjection, "visibleTurnItems">,
131): string {
132 return projection.visibleTurnItems
133 .flatMap((row) => {
134 const item = row.item;
135 return item.type === "user_message" || item.type === "assistant_message" ? [item.text] : [];
136 })
137 .join("\n");
138}
139
140describe("orchestration V2 thread fork", () => {
141 it.effect(
142 "creates an idle app fork and resolves it with Claude native session fork on first dispatch",
143 () =>
144 Effect.gen(function* () {
145 const rawTranscript = yield* readTranscript(CLAUDE_TRANSCRIPT_PATH);
146 const transcript = yield* ClaudeOrchestratorReplayHarness.decodeTranscript(rawTranscript);
147 const forkedNativeSessionId = metadataString(transcript, "forkedNativeSessionId");
148 const cwd = yield* Effect.acquireRelease(makeCheckpointWorkspace, (directory) =>
149 Effect.service(FileSystem.FileSystem).pipe(
150 Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })),
151 Effect.orDie,
152 ),
153 );
154
155 const materialized = yield* Effect.gen(function* () {
156 const ids = yield* IdAllocator.IdAllocatorV2;
157 const projectId = yield* ids.allocate.project({ fixtureName: "thread-fork-native" });
158 const sourceThreadId = yield* ids.allocate.thread({
159 fixtureName: "thread-fork-native-source",
160 projectId,
161 });
162 const targetThreadId = ThreadId.make("thread-fork-native-target");
163
164 const commands = [
165 {
166 type: "thread.create",
167 createdBy: "user",
168 creationSource: "web",
169 commandId: yield* ids.allocate.command({
170 fixtureName: "thread-fork-native",
171 commandName: "thread-create-source",
172 }),
173 threadId: sourceThreadId,
174 projectId,
175 title: "Source thread",
176 modelSelection: CLAUDE_MODEL_SELECTION,
177 runtimeMode: "full-access",
178 interactionMode: "default",
179 branch: null,
180 worktreePath: null,
181 },
182 {
183 type: "message.dispatch",
184 createdBy: "user",
185 creationSource: "web",
186 commandId: yield* ids.allocate.command({
187 fixtureName: "thread-fork-native",
188 commandName: "source-message",
189 }),
190 threadId: sourceThreadId,
191 messageId: MessageId.make("message-thread-fork-native-source"),
192 text: THREAD_FORK_NATIVE_SOURCE_PROMPT,
193 attachments: [],
194 modelSelection: CLAUDE_MODEL_SELECTION,
195 dispatchMode: { type: "start_immediately" },
196 },
197 {
198 type: "thread.fork",
199 createdBy: "user",
200 creationSource: "web",
201 commandId: CommandId.make("command-thread-fork-native"),
202 sourceThreadId,
203 targetThreadId,
204 sourcePoint: { type: "latest_stable" },
205 title: "Forked thread",
206 },
207 {
208 type: "message.dispatch",
209 createdBy: "user",
210 creationSource: "web",
211 commandId: yield* ids.allocate.command({
212 fixtureName: "thread-fork-native",
213 commandName: "target-message",
214 }),
215 threadId: targetThreadId,
216 messageId: MessageId.make("message-thread-fork-native-target"),
217 text: THREAD_FORK_NATIVE_TARGET_PROMPT,
218 attachments: [],
219 modelSelection: CLAUDE_MODEL_SELECTION,
220 dispatchMode: { type: "start_immediately" },
221 },
222 ] satisfies ReadonlyArray<OrchestrationV2Command>;
223
224 return {
225 sourceThreadId,
226 targetThreadId,
227 commands,
228 };
229 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
230
231 const result = yield* runOrchestratorV2ProviderReplayScenario(
232 {
233 name: "thread_fork_native/claude",
234 transcript,
235 commands: materialized.commands,
236 steps: [
237 { type: "dispatch", command: materialized.commands[0]!, await: true },
238 { type: "advance_clock", duration: "1 millis" },
239 { type: "dispatch", command: materialized.commands[1]!, await: true },
240 { type: "await_thread_idle", threadId: materialized.sourceThreadId },
241 { type: "dispatch", command: materialized.commands[2]!, await: true },
242 { type: "dispatch", command: materialized.commands[3]!, await: true },
243 { type: "await_thread_idle", threadId: materialized.targetThreadId },
244 ],
245 projectionThreadIds: [materialized.sourceThreadId, materialized.targetThreadId],
246 runtimePolicyOverride: { cwd },
247 },
248 ClaudeOrchestratorReplayHarness,
249 ).pipe(provideDeterministicTestRuntime);
250
251 const sourceProjection = result.projections.get(materialized.sourceThreadId);
252 const targetProjection = result.projections.get(materialized.targetThreadId);
253 assert.isDefined(sourceProjection);
254 assert.isDefined(targetProjection);
255 assert.equal(
256 targetProjection.providerThreads[0]?.nativeThreadRef?.nativeId,
257 forkedNativeSessionId,
258 );
259 assert.equal(
260 targetProjection.providerThreads[0]?.forkedFrom?.providerThreadId,
261 sourceProjection.providerThreads[0]?.id,
262 );
263 assert.include(
264 targetProjection.turnItems
265 .filter((item) => item.type === "assistant_message")
266 .map((item) => item.text)
267 .join("\n"),
268 "fork native ok",
269 );
270
271 const transfers = targetProjection.contextTransfers.filter(
272 (transfer) => transfer.targetThreadId === materialized.targetThreadId,
273 );
274 assert.lengthOf(transfers, 1);
275 assert.equal(transfers[0]?.status, "consumed");
276 assert.equal(transfers[0]?.resolution?.strategy, "native_fork");
277 const sourceUsage = sourceProjection.providerTurns[0]?.tokenUsage;
278 assert.isDefined(sourceUsage);
279 assert.deepEqual(transfers[0]?.sourceTokenUsage, sourceUsage);
280 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
281 );
282
283 it.effect("forks a Codex native thread at the selected native turn boundary", () =>
284 Effect.gen(function* () {
285 const rawTranscript = yield* readTranscript(PRIOR_TURN_TRANSCRIPT_PATH);
286 const cwd = yield* Effect.acquireRelease(makeCheckpointWorkspace, (directory) =>
287 Effect.service(FileSystem.FileSystem).pipe(
288 Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })),
289 Effect.orDie,
290 ),
291 );
292 const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript(
293 materializeReplayTranscriptWorkspace(rawTranscript, cwd),
294 );
295
296 const materialized = yield* Effect.gen(function* () {
297 const ids = yield* IdAllocator.IdAllocatorV2;
298 const projectId = yield* ids.allocate.project({
299 fixtureName: "thread-fork-native-prior-turn",
300 });
301 const sourceThreadId = yield* ids.allocate.thread({
302 fixtureName: "thread-fork-native-prior-turn-source",
303 projectId,
304 });
305 const targetThreadId = ThreadId.make("thread-fork-native-prior-turn-target");
306 const firstRunId = ids.derive.run({ threadId: sourceThreadId, ordinal: 1 });
307
308 const commands = [
309 {
310 type: "thread.create",
311 createdBy: "user",
312 creationSource: "web",
313 commandId: yield* ids.allocate.command({
314 fixtureName: "thread-fork-native-prior-turn",
315 commandName: "thread-create-source",
316 }),
317 threadId: sourceThreadId,
318 projectId,
319 title: "Source thread",
320 modelSelection: CODEX_MODEL_SELECTION,
321 runtimeMode: "full-access",
322 interactionMode: "default",
323 branch: null,
324 worktreePath: null,
325 },
326 {
327 type: "message.dispatch",
328 createdBy: "user",
329 creationSource: "web",
330 commandId: yield* ids.allocate.command({
331 fixtureName: "thread-fork-native-prior-turn",
332 commandName: "source-message-alpha",
333 }),
334 threadId: sourceThreadId,
335 messageId: MessageId.make("message-thread-fork-native-prior-turn-alpha"),
336 text: THREAD_FORK_NATIVE_PRIOR_TURN_ALPHA_PROMPT,
337 attachments: [],
338 modelSelection: CODEX_MODEL_SELECTION,
339 dispatchMode: { type: "start_immediately" },
340 },
341 {
342 type: "message.dispatch",
343 createdBy: "user",
344 creationSource: "web",
345 commandId: yield* ids.allocate.command({
346 fixtureName: "thread-fork-native-prior-turn",
347 commandName: "source-message-beta",
348 }),
349 threadId: sourceThreadId,
350 messageId: MessageId.make("message-thread-fork-native-prior-turn-beta"),
351 text: THREAD_FORK_NATIVE_PRIOR_TURN_BETA_PROMPT,
352 attachments: [],
353 modelSelection: CODEX_MODEL_SELECTION,
354 dispatchMode: { type: "start_immediately" },
355 },
356 {
357 type: "thread.fork",
358 createdBy: "user",
359 creationSource: "web",
360 commandId: CommandId.make("command-thread-fork-native-prior-turn"),
361 sourceThreadId,
362 targetThreadId,
363 sourcePoint: { type: "run", runId: firstRunId },
364 title: "Forked from first response",
365 },
366 {
367 type: "message.dispatch",
368 createdBy: "user",
369 creationSource: "web",
370 commandId: yield* ids.allocate.command({
371 fixtureName: "thread-fork-native-prior-turn",
372 commandName: "target-message-repeat",
373 }),
374 threadId: targetThreadId,
375 messageId: MessageId.make("message-thread-fork-native-prior-turn-repeat"),
376 text: THREAD_FORK_NATIVE_PRIOR_TURN_REPEAT_PROMPT,
377 attachments: [],
378 modelSelection: CODEX_MODEL_SELECTION,
379 dispatchMode: { type: "start_immediately" },
380 },
381 ] satisfies ReadonlyArray<OrchestrationV2Command>;
382
383 return {
384 sourceThreadId,
385 targetThreadId,
386 commands,
387 };
388 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
389
390 const result = yield* runOrchestratorV2ProviderReplayScenario(
391 {
392 name: "thread_fork_native_prior_turn/codex",
393 transcript,
394 commands: materialized.commands,
395 steps: [
396 { type: "dispatch", command: materialized.commands[0]!, await: true },
397 { type: "advance_clock", duration: "1 millis" },
398 { type: "dispatch", command: materialized.commands[1]!, await: true },
399 { type: "await_thread_idle", threadId: materialized.sourceThreadId },
400 { type: "dispatch", command: materialized.commands[2]!, await: true },
401 { type: "await_thread_idle", threadId: materialized.sourceThreadId },
402 { type: "dispatch", command: materialized.commands[3]!, await: true },
403 { type: "dispatch", command: materialized.commands[4]!, await: true },
404 { type: "await_thread_idle", threadId: materialized.targetThreadId },
405 ],
406 projectionThreadIds: [materialized.sourceThreadId, materialized.targetThreadId],
407 runtimePolicyOverride: { cwd, ...CODEX_READ_ONLY_NEVER_POLICY },
408 },
409 CodexOrchestratorReplayHarness,
410 ).pipe(provideDeterministicTestRuntime);
411
412 const targetProjection = result.projections.get(materialized.targetThreadId);
413 assert.isDefined(targetProjection);
414 const targetAssistantText = targetProjection.turnItems
415 .filter((item) => item.type === "assistant_message")
416 .map((item) => item.text)
417 .join("\n");
418 assert.include(targetAssistantText, "fork boundary alpha");
419 assert.notInclude(
420 targetAssistantText,
421 "fork boundary beta",
422 "forking from the first source run must not preserve later source turns in native Codex context",
423 );
424 assert.equal(targetProjection.contextTransfers[0]?.resolution?.strategy, "native_fork");
425
426 const visibleItems = targetProjection.visibleTurnItems.map((row) => row.item);
427 assert.deepEqual(
428 visibleItems.slice(0, 2).map((item) => item.type),
429 ["user_message", "assistant_message"],
430 "fork target projection should expose inherited source history through the fork point",
431 );
432 assert.equal(
433 visibleItems[0]?.type === "user_message" ? visibleItems[0].inputIntent : undefined,
434 "turn_start",
435 "inherited fork history should preserve source message intent",
436 );
437 assert.equal(targetProjection.visibleTurnItems[0]?.visibility, "inherited");
438 assert.equal(targetProjection.visibleTurnItems[1]?.visibility, "inherited");
439 const forkMarker = targetProjection.visibleTurnItems.find((row) => row.item.type === "fork");
440 assert.isDefined(forkMarker, "fork target projection should include a visible fork marker");
441 assert.equal(forkMarker.visibility, "synthetic");
442 const targetShell = result.shellSnapshot.threads.find(
443 (thread) => thread.id === materialized.targetThreadId,
444 );
445 assert.isDefined(targetShell, "shell snapshot should include the fork target thread");
446 assert.equal(targetShell.visibleItemCount, targetProjection.visibleTurnItems.length);
447 assert.equal(targetShell.lineage.relationshipToParent, "fork");
448 assert.equal(targetShell.forkedFrom?.type, "run");
449
450 const visibleText = visibleItems
451 .filter((item) => item.type === "user_message" || item.type === "assistant_message")
452 .map((item) => item.text)
453 .join("\n");
454 assert.include(visibleText, "fork boundary alpha");
455 assert.notInclude(
456 visibleText,
457 "fork boundary beta",
458 "fork target visible projection must not inherit source turns after the fork point",
459 );
460 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
461 );
462
463 it.effect("keeps a Claude native fork stable when the source thread rolls back", () =>
464 Effect.gen(function* () {
465 const rawTranscript = yield* readTranscript(CLAUDE_PRIOR_TURN_TRANSCRIPT_PATH);
466 const transcript = yield* ClaudeOrchestratorReplayHarness.decodeTranscript(rawTranscript);
467 const forkedNativeSessionId = metadataString(transcript, "forkedNativeSessionId");
468 const sourceAssistantMessageUuids = metadataStringArray(
469 transcript,
470 "sourceAssistantMessageUuids",
471 );
472 const cwd = yield* Effect.acquireRelease(makeCheckpointWorkspace, (directory) =>
473 Effect.service(FileSystem.FileSystem).pipe(
474 Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })),
475 Effect.orDie,
476 ),
477 );
478
479 const materialized = yield* Effect.gen(function* () {
480 const ids = yield* IdAllocator.IdAllocatorV2;
481 const projectId = yield* ids.allocate.project({
482 fixtureName: "thread-fork-native-prior-turn-source-rollback",
483 });
484 const sourceThreadId = yield* ids.allocate.thread({
485 fixtureName: "thread-fork-native-prior-turn-source-rollback-source",
486 projectId,
487 });
488 const targetThreadId = ThreadId.make(
489 "thread-fork-native-prior-turn-source-rollback-target",
490 );
491 const firstRunId = ids.derive.run({ threadId: sourceThreadId, ordinal: 1 });
492 const secondRunId = ids.derive.run({ threadId: sourceThreadId, ordinal: 2 });
493 const checkpointScopeId = yield* ids.allocate.checkpointScope({
494 threadId: sourceThreadId,
495 name: "root",
496 });
497 const firstCheckpointId = yield* ids.allocate.checkpoint({
498 checkpointScopeId,
499 name: "1",
500 });
501
502 const commands = [
503 {
504 type: "thread.create",
505 createdBy: "user",
506 creationSource: "web",
507 commandId: yield* ids.allocate.command({
508 fixtureName: "thread-fork-native-prior-turn-source-rollback",
509 commandName: "thread-create-source",
510 }),
511 threadId: sourceThreadId,
512 projectId,
513 title: "Source thread",
514 modelSelection: CLAUDE_MODEL_SELECTION,
515 runtimeMode: "full-access",
516 interactionMode: "default",
517 branch: null,
518 worktreePath: null,
519 },
520 {
521 type: "message.dispatch",
522 createdBy: "user",
523 creationSource: "web",
524 commandId: yield* ids.allocate.command({
525 fixtureName: "thread-fork-native-prior-turn-source-rollback",
526 commandName: "source-message-alpha",
527 }),
528 threadId: sourceThreadId,
529 messageId: MessageId.make(
530 "message-thread-fork-native-prior-turn-source-rollback-alpha",
531 ),
532 text: THREAD_FORK_NATIVE_PRIOR_TURN_ALPHA_PROMPT,
533 attachments: [],
534 modelSelection: CLAUDE_MODEL_SELECTION,
535 dispatchMode: { type: "start_immediately" },
536 },
537 {
538 type: "message.dispatch",
539 createdBy: "user",
540 creationSource: "web",
541 commandId: yield* ids.allocate.command({
542 fixtureName: "thread-fork-native-prior-turn-source-rollback",
543 commandName: "source-message-beta",
544 }),
545 threadId: sourceThreadId,
546 messageId: MessageId.make("message-thread-fork-native-prior-turn-source-rollback-beta"),
547 text: THREAD_FORK_NATIVE_PRIOR_TURN_BETA_PROMPT,
548 attachments: [],
549 modelSelection: CLAUDE_MODEL_SELECTION,
550 dispatchMode: { type: "start_immediately" },
551 },
552 {
553 type: "thread.fork",
554 createdBy: "user",
555 creationSource: "web",
556 commandId: CommandId.make("command-thread-fork-native-prior-turn-source-rollback"),
557 sourceThreadId,
558 targetThreadId,
559 sourcePoint: { type: "run", runId: firstRunId },
560 title: "Forked from first response",
561 },
562 {
563 type: "message.dispatch",
564 createdBy: "user",
565 creationSource: "web",
566 commandId: yield* ids.allocate.command({
567 fixtureName: "thread-fork-native-prior-turn-source-rollback",
568 commandName: "target-message-repeat",
569 }),
570 threadId: targetThreadId,
571 messageId: MessageId.make(
572 "message-thread-fork-native-prior-turn-source-rollback-repeat",
573 ),
574 text: THREAD_FORK_NATIVE_PRIOR_TURN_REPEAT_PROMPT,
575 attachments: [],
576 modelSelection: CLAUDE_MODEL_SELECTION,
577 dispatchMode: { type: "start_immediately" },
578 },
579 {
580 type: "checkpoint.rollback",
581 restoreFiles: false,
582 commandId: yield* ids.allocate.command({
583 fixtureName: "thread-fork-native-prior-turn-source-rollback",
584 commandName: "rollback-source-to-alpha",
585 }),
586 threadId: sourceThreadId,
587 scopeId: checkpointScopeId,
588 checkpointId: firstCheckpointId,
589 },
590 ] satisfies ReadonlyArray<OrchestrationV2Command>;
591
592 return {
593 sourceThreadId,
594 targetThreadId,
595 secondRunId,
596 commands,
597 };
598 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
599
600 const result = yield* runOrchestratorV2ProviderReplayScenario(
601 {
602 name: "thread_fork_native_prior_turn_source_rollback/claude",
603 transcript,
604 commands: materialized.commands,
605 steps: [
606 { type: "dispatch", command: materialized.commands[0]!, await: true },
607 { type: "advance_clock", duration: "1 millis" },
608 { type: "dispatch", command: materialized.commands[1]!, await: true },
609 { type: "await_thread_idle", threadId: materialized.sourceThreadId },
610 { type: "dispatch", command: materialized.commands[2]!, await: true },
611 { type: "await_thread_idle", threadId: materialized.sourceThreadId },
612 { type: "dispatch", command: materialized.commands[3]!, await: true },
613 { type: "dispatch", command: materialized.commands[4]!, await: true },
614 { type: "await_thread_idle", threadId: materialized.targetThreadId },
615 { type: "dispatch", command: materialized.commands[5]!, await: true },
616 {
617 type: "await_run_status",
618 threadId: materialized.sourceThreadId,
619 runId: materialized.secondRunId,
620 status: "rolled_back",
621 },
622 ],
623 projectionThreadIds: [materialized.sourceThreadId, materialized.targetThreadId],
624 runtimePolicyOverride: { cwd },
625 },
626 ClaudeOrchestratorReplayHarness,
627 ).pipe(provideDeterministicTestRuntime);
628
629 const sourceProjection = result.projections.get(materialized.sourceThreadId);
630 const targetProjection = result.projections.get(materialized.targetThreadId);
631 assert.isDefined(sourceProjection);
632 assert.isDefined(targetProjection);
633
634 assert.equal(
635 sourceProjection.runs.map((run) => run.status).join(","),
636 "completed,rolled_back",
637 );
638 assert.equal(
639 sourceProjection.providerThreads[0]?.nativeConversationHeadRef?.nativeId,
640 sourceAssistantMessageUuids[0],
641 "source rollback should persist the Claude resume cursor for the first assistant message",
642 );
643 assert.equal(
644 targetProjection.providerThreads[0]?.nativeThreadRef?.nativeId,
645 forkedNativeSessionId,
646 );
647 assert.equal(targetProjection.providerThreads[0]?.nativeConversationHeadRef, null);
648
649 const sourceVisibleText = userAndAssistantText(sourceProjection);
650 assert.include(sourceVisibleText, "fork boundary alpha");
651 assert.notInclude(sourceVisibleText, "fork boundary beta");
652
653 const targetVisibleText = userAndAssistantText(targetProjection);
654 assert.include(targetVisibleText, "fork boundary alpha");
655 assert.notInclude(
656 targetVisibleText,
657 "fork boundary beta",
658 "source rollback must not cause the fork target to inherit turns past its fork point",
659 );
660 assert.equal(targetProjection.contextTransfers[0]?.resolution?.strategy, "native_fork");
661 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
662 );
663
664 it.effect("rolls back a Claude native fork to an earlier fork-local turn", () =>
665 Effect.gen(function* () {
666 const rawTranscript = yield* readTranscript(CLAUDE_FORK_LOCAL_ROLLBACK_TRANSCRIPT_PATH);
667 const transcript = yield* ClaudeOrchestratorReplayHarness.decodeTranscript(rawTranscript);
668 const forkedNativeSessionId = metadataString(transcript, "forkedNativeSessionId");
669 const resumeSessionAt = metadataString(transcript, "resumeSessionAt");
670 const prompts = metadataStringArray(transcript, "prompts");
671 const [sourcePrompt, forkFirstPrompt, forkSecondPrompt, repeatPrompt] = prompts;
672 if (
673 sourcePrompt === undefined ||
674 forkFirstPrompt === undefined ||
675 forkSecondPrompt === undefined ||
676 repeatPrompt === undefined
677 ) {
678 throw new Error("Claude fork-local rollback transcript is missing expected prompts.");
679 }
680
681 const cwd = yield* Effect.acquireRelease(makeCheckpointWorkspace, (directory) =>
682 Effect.service(FileSystem.FileSystem).pipe(
683 Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })),
684 Effect.orDie,
685 ),
686 );
687
688 const materialized = yield* Effect.gen(function* () {
689 const ids = yield* IdAllocator.IdAllocatorV2;
690 const projectId = yield* ids.allocate.project({
691 fixtureName: "thread-fork-native-fork-local-rollback",
692 });
693 const sourceThreadId = yield* ids.allocate.thread({
694 fixtureName: "thread-fork-native-fork-local-rollback-source",
695 projectId,
696 });
697 const targetThreadId = ThreadId.make("thread-fork-native-fork-local-rollback-target");
698 const targetSecondRunId = ids.derive.run({ threadId: targetThreadId, ordinal: 2 });
699 const targetCheckpointScopeId = yield* ids.allocate.checkpointScope({
700 threadId: targetThreadId,
701 name: "root",
702 });
703 const targetFirstCheckpointId = yield* ids.allocate.checkpoint({
704 checkpointScopeId: targetCheckpointScopeId,
705 name: "1",
706 });
707
708 const commands = [
709 {
710 type: "thread.create",
711 createdBy: "user",
712 creationSource: "web",
713 commandId: yield* ids.allocate.command({
714 fixtureName: "thread-fork-native-fork-local-rollback",
715 commandName: "thread-create-source",
716 }),
717 threadId: sourceThreadId,
718 projectId,
719 title: "Source thread",
720 modelSelection: CLAUDE_MODEL_SELECTION,
721 runtimeMode: "full-access",
722 interactionMode: "default",
723 branch: null,
724 worktreePath: null,
725 },
726 {
727 type: "message.dispatch",
728 createdBy: "user",
729 creationSource: "web",
730 commandId: yield* ids.allocate.command({
731 fixtureName: "thread-fork-native-fork-local-rollback",
732 commandName: "source-message",
733 }),
734 threadId: sourceThreadId,
735 messageId: MessageId.make("message-thread-fork-native-fork-local-rollback-source"),
736 text: sourcePrompt,
737 attachments: [],
738 modelSelection: CLAUDE_MODEL_SELECTION,
739 dispatchMode: { type: "start_immediately" },
740 },
741 {
742 type: "thread.fork",
743 createdBy: "user",
744 creationSource: "web",
745 commandId: CommandId.make("command-thread-fork-native-fork-local-rollback"),
746 sourceThreadId,
747 targetThreadId,
748 sourcePoint: { type: "latest_stable" },
749 title: "Forked thread",
750 },
751 {
752 type: "message.dispatch",
753 createdBy: "user",
754 creationSource: "web",
755 commandId: yield* ids.allocate.command({
756 fixtureName: "thread-fork-native-fork-local-rollback",
757 commandName: "fork-first-message",
758 }),
759 threadId: targetThreadId,
760 messageId: MessageId.make("message-thread-fork-native-fork-local-rollback-first"),
761 text: forkFirstPrompt,
762 attachments: [],
763 modelSelection: CLAUDE_MODEL_SELECTION,
764 dispatchMode: { type: "start_immediately" },
765 },
766 {
767 type: "message.dispatch",
768 createdBy: "user",
769 creationSource: "web",
770 commandId: yield* ids.allocate.command({
771 fixtureName: "thread-fork-native-fork-local-rollback",
772 commandName: "fork-second-message",
773 }),
774 threadId: targetThreadId,
775 messageId: MessageId.make("message-thread-fork-native-fork-local-rollback-second"),
776 text: forkSecondPrompt,
777 attachments: [],
778 modelSelection: CLAUDE_MODEL_SELECTION,
779 dispatchMode: { type: "start_immediately" },
780 },
781 {
782 type: "checkpoint.rollback",
783 restoreFiles: false,
784 commandId: yield* ids.allocate.command({
785 fixtureName: "thread-fork-native-fork-local-rollback",
786 commandName: "rollback-fork-to-first",
787 }),
788 threadId: targetThreadId,
789 scopeId: targetCheckpointScopeId,
790 checkpointId: targetFirstCheckpointId,
791 },
792 {
793 type: "message.dispatch",
794 createdBy: "user",
795 creationSource: "web",
796 commandId: yield* ids.allocate.command({
797 fixtureName: "thread-fork-native-fork-local-rollback",
798 commandName: "fork-repeat-after-rollback",
799 }),
800 threadId: targetThreadId,
801 messageId: MessageId.make("message-thread-fork-native-fork-local-rollback-repeat"),
802 text: repeatPrompt,
803 attachments: [],
804 modelSelection: CLAUDE_MODEL_SELECTION,
805 dispatchMode: { type: "start_immediately" },
806 },
807 ] satisfies ReadonlyArray<OrchestrationV2Command>;
808
809 return {
810 sourceThreadId,
811 targetThreadId,
812 targetSecondRunId,
813 commands,
814 };
815 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
816
817 const result = yield* runOrchestratorV2ProviderReplayScenario(
818 {
819 name: "thread_fork_native_fork_local_rollback/claude",
820 transcript,
821 commands: materialized.commands,
822 steps: [
823 { type: "dispatch", command: materialized.commands[0]!, await: true },
824 { type: "advance_clock", duration: "1 millis" },
825 { type: "dispatch", command: materialized.commands[1]!, await: true },
826 { type: "await_thread_idle", threadId: materialized.sourceThreadId },
827 { type: "dispatch", command: materialized.commands[2]!, await: true },
828 { type: "dispatch", command: materialized.commands[3]!, await: true },
829 { type: "await_thread_idle", threadId: materialized.targetThreadId },
830 { type: "dispatch", command: materialized.commands[4]!, await: true },
831 { type: "await_thread_idle", threadId: materialized.targetThreadId },
832 { type: "dispatch", command: materialized.commands[5]!, await: true },
833 {
834 type: "await_run_status",
835 threadId: materialized.targetThreadId,
836 runId: materialized.targetSecondRunId,
837 status: "rolled_back",
838 },
839 { type: "dispatch", command: materialized.commands[6]!, await: true },
840 { type: "await_thread_idle", threadId: materialized.targetThreadId },
841 ],
842 projectionThreadIds: [materialized.sourceThreadId, materialized.targetThreadId],
843 runtimePolicyOverride: { cwd },
844 },
845 ClaudeOrchestratorReplayHarness,
846 ).pipe(provideDeterministicTestRuntime);
847
848 const targetProjection = result.projections.get(materialized.targetThreadId);
849 assert.isDefined(targetProjection);
850 assert.equal(
851 targetProjection.providerThreads[0]?.nativeThreadRef?.nativeId,
852 forkedNativeSessionId,
853 );
854 assert.isString(resumeSessionAt);
855
856 const targetVisibleText = userAndAssistantText(targetProjection);
857 assert.include(targetVisibleText, "fork local source alpha");
858 assert.include(targetVisibleText, "fork local first");
859 assert.notInclude(
860 targetVisibleText,
861 "fork local second",
862 "rolled back fork-local turns must disappear from the projected fork thread",
863 );
864
865 const targetVisibleAssistantText = targetProjection.visibleTurnItems
866 .map((row) => row.item)
867 .filter((item) => item.type === "assistant_message")
868 .map((item) => item.text)
869 .join("\n");
870 assert.notInclude(
871 targetVisibleAssistantText,
872 "fork local second",
873 "resumeSessionAt must reopen the Claude fork before the rolled-back second fork turn",
874 );
875 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
876 );
877});