1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import {
4 CommandId,
5 MessageId,
6 type ModelSelection,
7 type OrchestrationV2Command,
8 type OrchestrationV2ThreadProjection,
9 ProjectId,
10 ProviderInstanceId,
11 ProviderDriverKind,
12 type ProviderReplayTranscript,
13 ThreadId,
14} from "@t3tools/contracts";
15import * as Effect from "effect/Effect";
16import * as FileSystem from "effect/FileSystem";
17import * as Layer from "effect/Layer";
18import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
19
20import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts";
21import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts";
22import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
23import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
24import {
25 CODEX_MODEL_SELECTION,
26 THREAD_MERGE_BACK_FORK_PROMPT,
27 THREAD_MERGE_BACK_HANDOFF_PROMPT,
28 THREAD_MERGE_BACK_RECALL,
29 THREAD_MERGE_BACK_RECALL_PROMPT,
30 THREAD_MERGE_BACK_SIBLINGS_FIRST_FORK_PROMPT,
31 THREAD_MERGE_BACK_SIBLINGS_FIRST_HANDOFF_PROMPT,
32 THREAD_MERGE_BACK_SIBLINGS_RECALL,
33 THREAD_MERGE_BACK_SIBLINGS_RECALL_PROMPT,
34 THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
35 THREAD_MERGE_BACK_SIBLINGS_SECOND_HANDOFF_PROMPT,
36 THREAD_MERGE_BACK_SIBLINGS_SOURCE_PROMPT,
37 THREAD_MERGE_BACK_SOURCE_PROMPT,
38} from "./fixtures/shared.ts";
39import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts";
40import { makeCheckpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
41import {
42 decodeProviderReplayNdjson,
43 materializeReplayTranscriptWorkspace,
44} from "@t3tools/provider-testing/replayTranscript";
45
46// These recorded 0.137 rollouts predate injection. Preserve their native fork
47// and turn exchanges, and assert the new history delivery at the adapter boundary.
48// Exact thread/inject_items frames are covered by CodexAdapterV2.test.ts.
49const CodexHistoryReplayHarness: typeof CodexOrchestratorReplayHarness = {
50 ...CodexOrchestratorReplayHarness,
51 makeProviderAdapterRegistryLayer: (transcript, options) =>
52 Layer.effect(
53 ProviderAdapterRegistry.ProviderAdapterRegistryV2,
54 Effect.gen(function* () {
55 const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
56 return {
57 ...registry,
58 get: (id) =>
59 registry.get(id).pipe(
60 Effect.map((adapter) => ({
61 ...adapter,
62 openSession: (input) =>
63 adapter.openSession(input).pipe(
64 Effect.map((session) => ({
65 ...session,
66 injectHistory: (history) =>
67 Effect.sync(() => {
68 const userTexts = history.messages
69 .filter((message) => message.role === "user")
70 .map((message) => message.text);
71 assert.lengthOf(userTexts, 1);
72 assert.include(
73 [
74 THREAD_MERGE_BACK_FORK_PROMPT,
75 THREAD_MERGE_BACK_SIBLINGS_FIRST_FORK_PROMPT,
76 THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
77 ],
78 userTexts[0],
79 );
80 assert.isTrue(
81 history.messages.some(
82 (message) =>
83 message.role === "assistant" && message.text.includes("stored"),
84 ),
85 );
86 return true;
87 }),
88 })),
89 ),
90 })),
91 ),
92 };
93 }),
94 ).pipe(
95 Layer.provide(
96 CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript, options),
97 ),
98 ),
99};
100
101const CLAUDE_MODEL_SELECTION = {
102 instanceId: ProviderInstanceId.make("claudeAgent"),
103 model: "claude-sonnet-4-6",
104} as const;
105const MERGE_BACK_USER_TEXT =
106 "Retain the transferred fork marker for later. Respond with exactly: merge delta stored";
107const FIRST_SIBLING_MERGE_USER_TEXT =
108 "Retain the first transferred marker for later. Respond with exactly: first merge delta stored";
109const SECOND_SIBLING_MERGE_USER_TEXT =
110 "Retain the second transferred marker for later. Respond with exactly: second merge delta stored";
111
112interface ProviderVariant {
113 readonly driver: ProviderDriverKind;
114 readonly modelSelection: ModelSelection;
115}
116
117const PROVIDERS: ReadonlyArray<ProviderVariant> = [
118 {
119 driver: ProviderDriverKind.make("codex"),
120 // Recorded on gpt-6-sol: gpt-6-luna tends to recall an acknowledgement instead of the markers.
121 modelSelection: { ...CODEX_MODEL_SELECTION, model: "gpt-6-sol" },
122 },
123 {
124 driver: ProviderDriverKind.make("claudeAgent"),
125 modelSelection: CLAUDE_MODEL_SELECTION,
126 },
127];
128
129function transcriptPath(scenario: string, driver: ProviderDriverKind): string {
130 const fileName = driver === "codex" ? "codex_transcript.ndjson" : "claude_transcript.ndjson";
131 return `${import.meta.dirname}/fixtures/${scenario}/${fileName}`;
132}
133
134function readTranscript(scenario: string, driver: ProviderDriverKind) {
135 return Effect.gen(function* () {
136 const fs = yield* FileSystem.FileSystem;
137 const text = yield* fs.readFileString(transcriptPath(scenario, driver));
138 return yield* decodeProviderReplayNdjson(text);
139 });
140}
141
142function compactExpectedText(text: string, maxLength = 240): string {
143 const compacted = text.replace(/\s+/gu, " ").trim();
144 return compacted.length <= maxLength ? compacted : `${compacted.slice(0, maxLength - 3)}...`;
145}
146
147function forkDeltaSummary(input: {
148 readonly sourceThreadId: ThreadId;
149 readonly targetThreadId: ThreadId;
150 readonly forkUserText: string;
151 readonly forkAssistantText: string;
152}): string {
153 return [
154 "Merge-back context from forked conversation.",
155 `Source thread: ${input.sourceThreadId}`,
156 `Target thread: ${input.targetThreadId}`,
157 "Covered fork runs: 1-1",
158 "",
159 "Fork delta:",
160 `- User: ${compactExpectedText(input.forkUserText)}`,
161 `- Assistant: ${compactExpectedText(input.forkAssistantText)}`,
162 "- Checkpoint: 0 files",
163 ].join("\n");
164}
165
166function providerMessage(summary: string, userText: string): string {
167 return [
168 "Context handoff (merge_back / fork_delta_summary):",
169 summary,
170 "",
171 "User message:",
172 userText,
173 ].join("\n");
174}
175
176function replaceExactString(value: unknown, from: string, to: string): unknown {
177 if (value === from) {
178 return to;
179 }
180 if (Array.isArray(value)) {
181 return value.map((entry) => replaceExactString(entry, from, to));
182 }
183 if (typeof value !== "object" || value === null) {
184 return value;
185 }
186 return Object.fromEntries(
187 Object.entries(value).map(([key, entry]) => [key, replaceExactString(entry, from, to)]),
188 );
189}
190
191function parameterizeHandoffs(
192 transcript: ProviderReplayTranscript,
193 replacements: ReadonlyArray<readonly [recorded: string, generated: string]>,
194): ProviderReplayTranscript {
195 return replacements.reduce(
196 (current, [recorded, generated]) =>
197 replaceExactString(current, recorded, generated) as ProviderReplayTranscript,
198 transcript,
199 );
200}
201
202function nativeSourceThreadId(transcript: ProviderReplayTranscript): string {
203 if (transcript.provider === "claudeAgent") {
204 const value = transcript.metadata?.nativeSessionId;
205 if (typeof value !== "string") {
206 throw new Error(`${transcript.scenario} is missing metadata.nativeSessionId.`);
207 }
208 return value;
209 }
210 for (const entry of transcript.entries) {
211 if (entry.type !== "emit_inbound") {
212 continue;
213 }
214 const frame = entry.frame as {
215 readonly method?: unknown;
216 readonly params?: { readonly thread?: { readonly id?: unknown } };
217 };
218 if (frame.method === "thread/started" && typeof frame.params?.thread?.id === "string") {
219 return frame.params.thread.id;
220 }
221 }
222 throw new Error(`${transcript.scenario} is missing the source native thread id.`);
223}
224
225function visibleConversationText(projection: OrchestrationV2ThreadProjection): string {
226 return projection.visibleTurnItems
227 .flatMap(({ item }) =>
228 item.type === "user_message" || item.type === "assistant_message" ? [item.text] : [],
229 )
230 .join("\n");
231}
232
233function normalizePipeSpacing(text: string): string {
234 return text.replace(/\s*\|\s*/gu, "|");
235}
236
237function makeCreateCommand(input: {
238 readonly commandId: CommandId;
239 readonly threadId: ThreadId;
240 readonly projectId: ProjectId;
241 readonly modelSelection: ModelSelection;
242}): OrchestrationV2Command {
243 return {
244 type: "thread.create",
245 createdBy: "user",
246 creationSource: "web",
247 commandId: input.commandId,
248 threadId: input.threadId,
249 projectId: input.projectId,
250 title: "Merge-back source",
251 modelSelection: input.modelSelection,
252 runtimeMode: "full-access",
253 interactionMode: "default",
254 branch: null,
255 worktreePath: null,
256 };
257}
258
259describe("orchestration V2 merge-back provider replay", () => {
260 it.effect.each(PROVIDERS)(
261 "merges one fork delta back into the original $driver thread",
262 (variant) =>
263 Effect.gen(function* () {
264 const rawTranscript = yield* readTranscript("thread_merge_back_continue", variant.driver);
265 const materialized = yield* Effect.gen(function* () {
266 const ids = yield* IdAllocator.IdAllocatorV2;
267 const projectId = yield* ids.allocate.project({
268 fixtureName: `thread-merge-back-${variant.driver}`,
269 });
270 const sourceThreadId = yield* ids.allocate.thread({
271 fixtureName: `thread-merge-back-${variant.driver}-source`,
272 projectId,
273 });
274 const forkThreadId = ThreadId.make(`thread-merge-back-${variant.driver}-fork`);
275 const forkRunId = ids.derive.run({ threadId: forkThreadId, ordinal: 1 });
276 const commands = [
277 makeCreateCommand({
278 commandId: yield* ids.allocate.command({
279 fixtureName: `thread-merge-back-${variant.driver}`,
280 commandName: "create",
281 }),
282 threadId: sourceThreadId,
283 projectId,
284 modelSelection: variant.modelSelection,
285 }),
286 {
287 type: "message.dispatch",
288 createdBy: "user",
289 creationSource: "web",
290 commandId: yield* ids.allocate.command({
291 fixtureName: `thread-merge-back-${variant.driver}`,
292 commandName: "source",
293 }),
294 threadId: sourceThreadId,
295 messageId: MessageId.make(`message-merge-source-${variant.driver}`),
296 text: THREAD_MERGE_BACK_SOURCE_PROMPT,
297 attachments: [],
298 modelSelection: variant.modelSelection,
299 dispatchMode: { type: "start_immediately" },
300 },
301 {
302 type: "thread.fork",
303 createdBy: "user",
304 creationSource: "web",
305 commandId: CommandId.make(`command-merge-fork-${variant.driver}`),
306 sourceThreadId,
307 targetThreadId: forkThreadId,
308 sourcePoint: { type: "latest_stable" },
309 title: "Merge-back fork",
310 },
311 {
312 type: "message.dispatch",
313 createdBy: "user",
314 creationSource: "web",
315 commandId: yield* ids.allocate.command({
316 fixtureName: `thread-merge-back-${variant.driver}`,
317 commandName: "fork-delta",
318 }),
319 threadId: forkThreadId,
320 messageId: MessageId.make(`message-merge-fork-${variant.driver}`),
321 text: THREAD_MERGE_BACK_FORK_PROMPT,
322 attachments: [],
323 modelSelection: variant.modelSelection,
324 dispatchMode: { type: "start_immediately" },
325 },
326 {
327 type: "thread.merge_back",
328 createdBy: "user",
329 creationSource: "web",
330 commandId: CommandId.make(`command-merge-back-${variant.driver}`),
331 sourceThreadId: forkThreadId,
332 targetThreadId: sourceThreadId,
333 sourcePoint: { type: "run", runId: forkRunId },
334 },
335 {
336 type: "message.dispatch",
337 createdBy: "user",
338 creationSource: "web",
339 commandId: yield* ids.allocate.command({
340 fixtureName: `thread-merge-back-${variant.driver}`,
341 commandName: "consume-merge",
342 }),
343 threadId: sourceThreadId,
344 messageId: MessageId.make(`message-consume-merge-${variant.driver}`),
345 text: MERGE_BACK_USER_TEXT,
346 attachments: [],
347 modelSelection: variant.modelSelection,
348 dispatchMode: { type: "start_immediately" },
349 },
350 {
351 type: "message.dispatch",
352 createdBy: "user",
353 creationSource: "web",
354 commandId: yield* ids.allocate.command({
355 fixtureName: `thread-merge-back-${variant.driver}`,
356 commandName: "recall",
357 }),
358 threadId: sourceThreadId,
359 messageId: MessageId.make(`message-merge-recall-${variant.driver}`),
360 text: THREAD_MERGE_BACK_RECALL_PROMPT,
361 attachments: [],
362 modelSelection: variant.modelSelection,
363 dispatchMode: { type: "start_immediately" },
364 },
365 ] satisfies ReadonlyArray<OrchestrationV2Command>;
366 return { commands, sourceThreadId, forkThreadId, forkRunId };
367 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
368 const summary = forkDeltaSummary({
369 sourceThreadId: materialized.forkThreadId,
370 targetThreadId: materialized.sourceThreadId,
371 forkUserText: THREAD_MERGE_BACK_FORK_PROMPT,
372 forkAssistantText: "merge fork stored",
373 });
374 const generatedProviderMessage = providerMessage(summary, MERGE_BACK_USER_TEXT);
375 const parameterizedTranscript = parameterizeHandoffs(rawTranscript, [
376 [
377 THREAD_MERGE_BACK_HANDOFF_PROMPT,
378 variant.driver === "codex" ? MERGE_BACK_USER_TEXT : generatedProviderMessage,
379 ],
380 ]);
381 const cwd = yield* Effect.acquireRelease(
382 Effect.promise(() => makeCheckpointWorkspace(`merge-back-${variant.driver}`)),
383 (directory) =>
384 Effect.service(FileSystem.FileSystem).pipe(
385 Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })),
386 Effect.orDie,
387 ),
388 );
389 const scenario = {
390 name: `thread_merge_back_continue/${variant.driver}`,
391 commands: materialized.commands,
392 steps: [
393 { type: "dispatch" as const, command: materialized.commands[0]!, await: true },
394 { type: "advance_clock" as const, duration: "1 millis" as const },
395 { type: "dispatch" as const, command: materialized.commands[1]!, await: true },
396 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
397 { type: "dispatch" as const, command: materialized.commands[2]!, await: true },
398 { type: "dispatch" as const, command: materialized.commands[3]!, await: true },
399 { type: "await_thread_idle" as const, threadId: materialized.forkThreadId },
400 { type: "dispatch" as const, command: materialized.commands[4]!, await: true },
401 { type: "dispatch" as const, command: materialized.commands[5]!, await: true },
402 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
403 { type: "dispatch" as const, command: materialized.commands[6]!, await: true },
404 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
405 ],
406 projectionThreadIds: [materialized.sourceThreadId, materialized.forkThreadId],
407 runtimePolicyOverride: { cwd },
408 };
409 const result =
410 variant.driver === "codex"
411 ? yield* runOrchestratorV2ProviderReplayScenario(
412 {
413 ...scenario,
414 transcript: yield* CodexOrchestratorReplayHarness.decodeTranscript(
415 materializeReplayTranscriptWorkspace(parameterizedTranscript, cwd),
416 ),
417 },
418 CodexHistoryReplayHarness,
419 ).pipe(provideDeterministicTestRuntime)
420 : yield* runOrchestratorV2ProviderReplayScenario(
421 {
422 ...scenario,
423 transcript:
424 yield* ClaudeOrchestratorReplayHarness.decodeTranscript(
425 parameterizedTranscript,
426 ),
427 },
428 ClaudeOrchestratorReplayHarness,
429 ).pipe(provideDeterministicTestRuntime);
430
431 const source = result.projections.get(materialized.sourceThreadId);
432 const fork = result.projections.get(materialized.forkThreadId);
433 assert.isDefined(source);
434 assert.isDefined(fork);
435 assert.equal(
436 source.providerThreads[0]?.nativeThreadRef?.nativeId,
437 nativeSourceThreadId(rawTranscript),
438 );
439 assert.lengthOf(source.contextHandoffs, 1);
440 assert.equal(source.contextHandoffs[0]?.summaryText, summary);
441 assert.equal(source.contextHandoffs[0]?.strategy, "fork_delta_summary");
442 const mergeTransfer = source.contextTransfers.find(
443 (transfer) => transfer.type === "merge_back",
444 );
445 assert.isDefined(mergeTransfer);
446 assert.equal(mergeTransfer.status, "consumed");
447 assert.equal(mergeTransfer.sourcePoint.runId, materialized.forkRunId);
448 assert.equal(mergeTransfer.resolution?.strategy, "fork_delta_context");
449 assert.include(
450 normalizePipeSpacing(visibleConversationText(source)),
451 THREAD_MERGE_BACK_RECALL,
452 );
453 assert.notInclude(visibleConversationText(source), "Context handoff (");
454 assert.include(visibleConversationText(fork), "merge fork stored");
455 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
456 );
457
458 it.effect.each(PROVIDERS)(
459 "merges two sibling fork deltas into the original $driver thread",
460 (variant) =>
461 Effect.gen(function* () {
462 const rawTranscript = yield* readTranscript("thread_merge_back_siblings", variant.driver);
463 const materialized = yield* Effect.gen(function* () {
464 const ids = yield* IdAllocator.IdAllocatorV2;
465 const fixtureName = `thread-merge-back-siblings-${variant.driver}`;
466 const projectId = yield* ids.allocate.project({ fixtureName });
467 const sourceThreadId = yield* ids.allocate.thread({
468 fixtureName: `${fixtureName}-source`,
469 projectId,
470 });
471 const firstForkThreadId = ThreadId.make(`${fixtureName}-first-fork`);
472 const secondForkThreadId = ThreadId.make(`${fixtureName}-second-fork`);
473 const firstForkRunId = ids.derive.run({ threadId: firstForkThreadId, ordinal: 1 });
474 const secondForkRunId = ids.derive.run({ threadId: secondForkThreadId, ordinal: 1 });
475 const commandId = (commandName: string) =>
476 ids.allocate.command({ fixtureName, commandName });
477 const commands = [
478 makeCreateCommand({
479 commandId: yield* commandId("create"),
480 threadId: sourceThreadId,
481 projectId,
482 modelSelection: variant.modelSelection,
483 }),
484 {
485 type: "message.dispatch",
486 createdBy: "user",
487 creationSource: "web",
488 commandId: yield* commandId("source"),
489 threadId: sourceThreadId,
490 messageId: MessageId.make(`message-${fixtureName}-source`),
491 text: THREAD_MERGE_BACK_SIBLINGS_SOURCE_PROMPT,
492 attachments: [],
493 modelSelection: variant.modelSelection,
494 dispatchMode: { type: "start_immediately" },
495 },
496 {
497 type: "thread.fork",
498 createdBy: "user",
499 creationSource: "web",
500 commandId: yield* commandId("first-fork"),
501 sourceThreadId,
502 targetThreadId: firstForkThreadId,
503 sourcePoint: { type: "latest_stable" },
504 title: "First merge-back sibling",
505 },
506 {
507 type: "message.dispatch",
508 createdBy: "user",
509 creationSource: "web",
510 commandId: yield* commandId("first-fork-delta"),
511 threadId: firstForkThreadId,
512 messageId: MessageId.make(`message-${fixtureName}-first-fork`),
513 text: THREAD_MERGE_BACK_SIBLINGS_FIRST_FORK_PROMPT,
514 attachments: [],
515 modelSelection: variant.modelSelection,
516 dispatchMode: { type: "start_immediately" },
517 },
518 {
519 type: "thread.fork",
520 createdBy: "user",
521 creationSource: "web",
522 commandId: yield* commandId("second-fork"),
523 sourceThreadId,
524 targetThreadId: secondForkThreadId,
525 sourcePoint: { type: "latest_stable" },
526 title: "Second merge-back sibling",
527 },
528 {
529 type: "message.dispatch",
530 createdBy: "user",
531 creationSource: "web",
532 commandId: yield* commandId("second-fork-delta"),
533 threadId: secondForkThreadId,
534 messageId: MessageId.make(`message-${fixtureName}-second-fork`),
535 text: THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
536 attachments: [],
537 modelSelection: variant.modelSelection,
538 dispatchMode: { type: "start_immediately" },
539 },
540 {
541 type: "thread.merge_back",
542 createdBy: "user",
543 creationSource: "web",
544 commandId: yield* commandId("merge-first"),
545 sourceThreadId: firstForkThreadId,
546 targetThreadId: sourceThreadId,
547 sourcePoint: { type: "run", runId: firstForkRunId },
548 },
549 {
550 type: "message.dispatch",
551 createdBy: "user",
552 creationSource: "web",
553 commandId: yield* commandId("consume-first"),
554 threadId: sourceThreadId,
555 messageId: MessageId.make(`message-${fixtureName}-consume-first`),
556 text: FIRST_SIBLING_MERGE_USER_TEXT,
557 attachments: [],
558 modelSelection: variant.modelSelection,
559 dispatchMode: { type: "start_immediately" },
560 },
561 {
562 type: "thread.merge_back",
563 createdBy: "user",
564 creationSource: "web",
565 commandId: yield* commandId("merge-second"),
566 sourceThreadId: secondForkThreadId,
567 targetThreadId: sourceThreadId,
568 sourcePoint: { type: "run", runId: secondForkRunId },
569 },
570 {
571 type: "message.dispatch",
572 createdBy: "user",
573 creationSource: "web",
574 commandId: yield* commandId("consume-second"),
575 threadId: sourceThreadId,
576 messageId: MessageId.make(`message-${fixtureName}-consume-second`),
577 text: SECOND_SIBLING_MERGE_USER_TEXT,
578 attachments: [],
579 modelSelection: variant.modelSelection,
580 dispatchMode: { type: "start_immediately" },
581 },
582 {
583 type: "message.dispatch",
584 createdBy: "user",
585 creationSource: "web",
586 commandId: yield* commandId("recall"),
587 threadId: sourceThreadId,
588 messageId: MessageId.make(`message-${fixtureName}-recall`),
589 text: THREAD_MERGE_BACK_SIBLINGS_RECALL_PROMPT,
590 attachments: [],
591 modelSelection: variant.modelSelection,
592 dispatchMode: { type: "start_immediately" },
593 },
594 ] satisfies ReadonlyArray<OrchestrationV2Command>;
595 return {
596 commands,
597 sourceThreadId,
598 firstForkThreadId,
599 secondForkThreadId,
600 firstForkRunId,
601 secondForkRunId,
602 };
603 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
604 const firstSummary = forkDeltaSummary({
605 sourceThreadId: materialized.firstForkThreadId,
606 targetThreadId: materialized.sourceThreadId,
607 forkUserText: THREAD_MERGE_BACK_SIBLINGS_FIRST_FORK_PROMPT,
608 forkAssistantText: "first merge sibling stored",
609 });
610 const secondSummary = forkDeltaSummary({
611 sourceThreadId: materialized.secondForkThreadId,
612 targetThreadId: materialized.sourceThreadId,
613 forkUserText: THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
614 forkAssistantText: "second merge sibling stored",
615 });
616 const transcript = parameterizeHandoffs(rawTranscript, [
617 [
618 THREAD_MERGE_BACK_SIBLINGS_FIRST_HANDOFF_PROMPT,
619 variant.driver === "codex"
620 ? FIRST_SIBLING_MERGE_USER_TEXT
621 : providerMessage(firstSummary, FIRST_SIBLING_MERGE_USER_TEXT),
622 ],
623 [
624 THREAD_MERGE_BACK_SIBLINGS_SECOND_HANDOFF_PROMPT,
625 variant.driver === "codex"
626 ? SECOND_SIBLING_MERGE_USER_TEXT
627 : providerMessage(secondSummary, SECOND_SIBLING_MERGE_USER_TEXT),
628 ],
629 ]);
630 const cwd = yield* Effect.acquireRelease(
631 Effect.promise(() => makeCheckpointWorkspace(`merge-back-siblings-${variant.driver}`)),
632 (directory) =>
633 Effect.service(FileSystem.FileSystem).pipe(
634 Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })),
635 Effect.orDie,
636 ),
637 );
638 const steps = [
639 { type: "dispatch" as const, command: materialized.commands[0]!, await: true },
640 { type: "advance_clock" as const, duration: "1 millis" as const },
641 { type: "dispatch" as const, command: materialized.commands[1]!, await: true },
642 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
643 { type: "dispatch" as const, command: materialized.commands[2]!, await: true },
644 { type: "dispatch" as const, command: materialized.commands[3]!, await: true },
645 { type: "await_thread_idle" as const, threadId: materialized.firstForkThreadId },
646 { type: "dispatch" as const, command: materialized.commands[4]!, await: true },
647 { type: "dispatch" as const, command: materialized.commands[5]!, await: true },
648 { type: "await_thread_idle" as const, threadId: materialized.secondForkThreadId },
649 { type: "dispatch" as const, command: materialized.commands[6]!, await: true },
650 { type: "dispatch" as const, command: materialized.commands[7]!, await: true },
651 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
652 { type: "dispatch" as const, command: materialized.commands[8]!, await: true },
653 { type: "dispatch" as const, command: materialized.commands[9]!, await: true },
654 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
655 { type: "dispatch" as const, command: materialized.commands[10]!, await: true },
656 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
657 ];
658 const scenario = {
659 name: `thread_merge_back_siblings/${variant.driver}`,
660 commands: materialized.commands,
661 steps,
662 projectionThreadIds: [
663 materialized.sourceThreadId,
664 materialized.firstForkThreadId,
665 materialized.secondForkThreadId,
666 ],
667 runtimePolicyOverride: { cwd },
668 };
669 const result =
670 variant.driver === "codex"
671 ? yield* runOrchestratorV2ProviderReplayScenario(
672 {
673 ...scenario,
674 transcript: yield* CodexOrchestratorReplayHarness.decodeTranscript(
675 materializeReplayTranscriptWorkspace(transcript, cwd),
676 ),
677 },
678 CodexHistoryReplayHarness,
679 ).pipe(provideDeterministicTestRuntime)
680 : yield* runOrchestratorV2ProviderReplayScenario(
681 {
682 ...scenario,
683 transcript: yield* ClaudeOrchestratorReplayHarness.decodeTranscript(transcript),
684 },
685 ClaudeOrchestratorReplayHarness,
686 ).pipe(provideDeterministicTestRuntime);
687
688 const source = result.projections.get(materialized.sourceThreadId);
689 const firstFork = result.projections.get(materialized.firstForkThreadId);
690 const secondFork = result.projections.get(materialized.secondForkThreadId);
691 assert.isDefined(source);
692 assert.isDefined(firstFork);
693 assert.isDefined(secondFork);
694 assert.equal(
695 source.providerThreads[0]?.nativeThreadRef?.nativeId,
696 nativeSourceThreadId(rawTranscript),
697 );
698 const firstForkNativeId = firstFork.providerThreads[0]?.nativeThreadRef?.nativeId;
699 const secondForkNativeId = secondFork.providerThreads[0]?.nativeThreadRef?.nativeId;
700 assert.isDefined(firstForkNativeId);
701 assert.isDefined(secondForkNativeId);
702 assert.notEqual(firstForkNativeId, secondForkNativeId);
703 assert.notEqual(firstForkNativeId, source.providerThreads[0]?.nativeThreadRef?.nativeId);
704 assert.notEqual(secondForkNativeId, source.providerThreads[0]?.nativeThreadRef?.nativeId);
705 assert.deepEqual(
706 source.contextHandoffs.map((handoff) => handoff.summaryText),
707 [firstSummary, secondSummary],
708 );
709 const mergeTransfers = source.contextTransfers.filter(
710 (transfer) => transfer.type === "merge_back",
711 );
712 assert.lengthOf(mergeTransfers, 2);
713 assert.deepEqual(
714 mergeTransfers.map((transfer) => transfer.status),
715 ["consumed", "consumed"],
716 );
717 assert.deepEqual(
718 mergeTransfers.map((transfer) => transfer.sourcePoint.runId),
719 [materialized.firstForkRunId, materialized.secondForkRunId],
720 );
721 assert.include(
722 normalizePipeSpacing(visibleConversationText(source)),
723 THREAD_MERGE_BACK_SIBLINGS_RECALL,
724 );
725 assert.notInclude(visibleConversationText(source), "Context handoff (");
726 assert.include(visibleConversationText(firstFork), "first merge sibling stored");
727 assert.notInclude(visibleConversationText(firstFork), "second merge sibling stored");
728 assert.include(visibleConversationText(secondFork), "second merge sibling stored");
729 assert.notInclude(visibleConversationText(secondFork), "first merge sibling stored");
730 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
731 );
732});