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_SIBLINGS_FIRST_FORK_PROMPT,
27 THREAD_MERGE_BACK_SIBLINGS_FIRST_HANDOFF_PROMPT,
28 THREAD_MERGE_BACK_SIBLINGS_RECALL,
29 THREAD_MERGE_BACK_SIBLINGS_RECALL_PROMPT,
30 THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
31 THREAD_MERGE_BACK_SIBLINGS_SECOND_HANDOFF_PROMPT,
32 THREAD_MERGE_BACK_SIBLINGS_SOURCE_PROMPT,
33} from "./fixtures/shared.ts";
34import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts";
35import { makeCheckpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
36import {
37 decodeProviderReplayNdjson,
38 materializeReplayTranscriptWorkspace,
39} from "@t3tools/provider-testing/replayTranscript";
40
41const CodexHistoryReplayHarness: typeof CodexOrchestratorReplayHarness = {
42 ...CodexOrchestratorReplayHarness,
43 makeProviderAdapterRegistryLayer: (transcript, options) =>
44 Layer.effect(
45 ProviderAdapterRegistry.ProviderAdapterRegistryV2,
46 Effect.gen(function* () {
47 const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
48 return {
49 ...registry,
50 get: (id) =>
51 registry.get(id).pipe(
52 Effect.map((adapter) => ({
53 ...adapter,
54 openSession: (input) =>
55 adapter.openSession(input).pipe(
56 Effect.map((session) => ({
57 ...session,
58 injectHistory: (history) =>
59 Effect.sync(() => {
60 const userTexts = history.messages
61 .filter((message) => message.role === "user")
62 .map((message) => message.text);
63 assert.lengthOf(userTexts, 1);
64 assert.include(
65 [
66 THREAD_MERGE_BACK_SIBLINGS_FIRST_FORK_PROMPT,
67 THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
68 ],
69 userTexts[0],
70 );
71 assert.isTrue(
72 history.messages.some(
73 (message) =>
74 message.role === "assistant" && message.text.includes("stored"),
75 ),
76 );
77 return true;
78 }),
79 })),
80 ),
81 })),
82 ),
83 };
84 }),
85 ).pipe(
86 Layer.provide(
87 CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript, options),
88 ),
89 ),
90};
91
92const CLAUDE_MODEL_SELECTION = {
93 instanceId: ProviderInstanceId.make("claudeAgent"),
94 model: "claude-sonnet-4-6",
95} as const;
96const FIRST_SIBLING_MERGE_USER_TEXT =
97 "Retain the first transferred marker for later. Respond with exactly: first merge delta stored";
98const SECOND_SIBLING_MERGE_USER_TEXT =
99 "Retain the second transferred marker for later. Respond with exactly: second merge delta stored";
100
101interface ProviderVariant {
102 readonly driver: ProviderDriverKind;
103 readonly modelSelection: ModelSelection;
104}
105
106const PROVIDERS: ReadonlyArray<ProviderVariant> = [
107 {
108 driver: ProviderDriverKind.make("codex"),
109 // Recorded on gpt-6-sol: gpt-6-luna tends to recall an acknowledgement instead of the markers.
110 modelSelection: { ...CODEX_MODEL_SELECTION, model: "gpt-6-sol" },
111 },
112 {
113 driver: ProviderDriverKind.make("claudeAgent"),
114 modelSelection: CLAUDE_MODEL_SELECTION,
115 },
116];
117
118function transcriptPath(scenario: string, driver: ProviderDriverKind): string {
119 const fileName = driver === "codex" ? "codex_transcript.ndjson" : "claude_transcript.ndjson";
120 return `${import.meta.dirname}/fixtures/${scenario}/${fileName}`;
121}
122
123function readTranscript(scenario: string, driver: ProviderDriverKind) {
124 return Effect.gen(function* () {
125 const fs = yield* FileSystem.FileSystem;
126 const text = yield* fs.readFileString(transcriptPath(scenario, driver));
127 return yield* decodeProviderReplayNdjson(text);
128 });
129}
130
131function compactExpectedText(text: string, maxLength = 240): string {
132 const compacted = text.replace(/\s+/gu, " ").trim();
133 return compacted.length <= maxLength ? compacted : `${compacted.slice(0, maxLength - 3)}...`;
134}
135
136function forkDeltaSummary(input: {
137 readonly sourceThreadId: ThreadId;
138 readonly targetThreadId: ThreadId;
139 readonly forkUserText: string;
140 readonly forkAssistantText: string;
141}): string {
142 return [
143 "Merge-back context from forked conversation.",
144 `Source thread: ${input.sourceThreadId}`,
145 `Target thread: ${input.targetThreadId}`,
146 "Covered fork runs: 1-1",
147 "",
148 "Fork delta:",
149 `- User: ${compactExpectedText(input.forkUserText)}`,
150 `- Assistant: ${compactExpectedText(input.forkAssistantText)}`,
151 "- Checkpoint: 0 files",
152 ].join("\n");
153}
154
155function providerMessage(summary: string, userText: string): string {
156 return [
157 "Context handoff (merge_back / fork_delta_summary):",
158 summary,
159 "",
160 "User message:",
161 userText,
162 ].join("\n");
163}
164
165function replaceExactString(value: unknown, from: string, to: string): unknown {
166 if (value === from) {
167 return to;
168 }
169 if (Array.isArray(value)) {
170 return value.map((entry) => replaceExactString(entry, from, to));
171 }
172 if (typeof value !== "object" || value === null) {
173 return value;
174 }
175 return Object.fromEntries(
176 Object.entries(value).map(([key, entry]) => [key, replaceExactString(entry, from, to)]),
177 );
178}
179
180function parameterizeHandoffs(
181 transcript: ProviderReplayTranscript,
182 replacements: ReadonlyArray<readonly [recorded: string, generated: string]>,
183): ProviderReplayTranscript {
184 return replacements.reduce(
185 (current, [recorded, generated]) =>
186 replaceExactString(current, recorded, generated) as ProviderReplayTranscript,
187 transcript,
188 );
189}
190
191function nativeSourceThreadId(transcript: ProviderReplayTranscript): string {
192 if (transcript.provider === "claudeAgent") {
193 const value = transcript.metadata?.nativeSessionId;
194 if (typeof value !== "string") {
195 throw new Error(`${transcript.scenario} is missing metadata.nativeSessionId.`);
196 }
197 return value;
198 }
199 for (const entry of transcript.entries) {
200 if (entry.type !== "emit_inbound") {
201 continue;
202 }
203 const frame = entry.frame as {
204 readonly method?: unknown;
205 readonly params?: { readonly thread?: { readonly id?: unknown } };
206 };
207 if (frame.method === "thread/started" && typeof frame.params?.thread?.id === "string") {
208 return frame.params.thread.id;
209 }
210 }
211 throw new Error(`${transcript.scenario} is missing the source native thread id.`);
212}
213
214function visibleConversationText(projection: OrchestrationV2ThreadProjection): string {
215 return projection.visibleTurnItems
216 .flatMap(({ item }) =>
217 item.type === "user_message" || item.type === "assistant_message" ? [item.text] : [],
218 )
219 .join("\n");
220}
221
222function normalizePipeSpacing(text: string): string {
223 return text.replace(/\s*\|\s*/gu, "|");
224}
225
226function makeCreateCommand(input: {
227 readonly commandId: CommandId;
228 readonly threadId: ThreadId;
229 readonly projectId: ProjectId;
230 readonly modelSelection: ModelSelection;
231}): OrchestrationV2Command {
232 return {
233 type: "thread.create",
234 createdBy: "user",
235 creationSource: "web",
236 commandId: input.commandId,
237 threadId: input.threadId,
238 projectId: input.projectId,
239 title: "Merge-back source",
240 modelSelection: input.modelSelection,
241 runtimeMode: "full-access",
242 interactionMode: "default",
243 branch: null,
244 worktreePath: null,
245 };
246}
247
248describe("orchestration V2 merge-back provider replay", () => {
249 it.effect.each(PROVIDERS)(
250 "merges two sibling fork deltas into the original $driver thread",
251 (variant) =>
252 Effect.gen(function* () {
253 const rawTranscript = yield* readTranscript("thread_merge_back_siblings", variant.driver);
254 const materialized = yield* Effect.gen(function* () {
255 const ids = yield* IdAllocator.IdAllocatorV2;
256 const fixtureName = `thread-merge-back-siblings-${variant.driver}`;
257 const projectId = yield* ids.allocate.project({ fixtureName });
258 const sourceThreadId = yield* ids.allocate.thread({
259 fixtureName: `${fixtureName}-source`,
260 projectId,
261 });
262 const firstForkThreadId = ThreadId.make(`${fixtureName}-first-fork`);
263 const secondForkThreadId = ThreadId.make(`${fixtureName}-second-fork`);
264 const firstForkRunId = ids.derive.run({ threadId: firstForkThreadId, ordinal: 1 });
265 const secondForkRunId = ids.derive.run({ threadId: secondForkThreadId, ordinal: 1 });
266 const commandId = (commandName: string) =>
267 ids.allocate.command({ fixtureName, commandName });
268 const commands = [
269 makeCreateCommand({
270 commandId: yield* commandId("create"),
271 threadId: sourceThreadId,
272 projectId,
273 modelSelection: variant.modelSelection,
274 }),
275 {
276 type: "message.dispatch",
277 createdBy: "user",
278 creationSource: "web",
279 commandId: yield* commandId("source"),
280 threadId: sourceThreadId,
281 messageId: MessageId.make(`message-${fixtureName}-source`),
282 text: THREAD_MERGE_BACK_SIBLINGS_SOURCE_PROMPT,
283 attachments: [],
284 modelSelection: variant.modelSelection,
285 dispatchMode: { type: "start_immediately" },
286 },
287 {
288 type: "thread.fork",
289 createdBy: "user",
290 creationSource: "web",
291 commandId: yield* commandId("first-fork"),
292 sourceThreadId,
293 targetThreadId: firstForkThreadId,
294 sourcePoint: { type: "latest_stable" },
295 title: "First merge-back sibling",
296 },
297 {
298 type: "message.dispatch",
299 createdBy: "user",
300 creationSource: "web",
301 commandId: yield* commandId("first-fork-delta"),
302 threadId: firstForkThreadId,
303 messageId: MessageId.make(`message-${fixtureName}-first-fork`),
304 text: THREAD_MERGE_BACK_SIBLINGS_FIRST_FORK_PROMPT,
305 attachments: [],
306 modelSelection: variant.modelSelection,
307 dispatchMode: { type: "start_immediately" },
308 },
309 {
310 type: "thread.fork",
311 createdBy: "user",
312 creationSource: "web",
313 commandId: yield* commandId("second-fork"),
314 sourceThreadId,
315 targetThreadId: secondForkThreadId,
316 sourcePoint: { type: "latest_stable" },
317 title: "Second merge-back sibling",
318 },
319 {
320 type: "message.dispatch",
321 createdBy: "user",
322 creationSource: "web",
323 commandId: yield* commandId("second-fork-delta"),
324 threadId: secondForkThreadId,
325 messageId: MessageId.make(`message-${fixtureName}-second-fork`),
326 text: THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
327 attachments: [],
328 modelSelection: variant.modelSelection,
329 dispatchMode: { type: "start_immediately" },
330 },
331 {
332 type: "thread.merge_back",
333 createdBy: "user",
334 creationSource: "web",
335 commandId: yield* commandId("merge-first"),
336 sourceThreadId: firstForkThreadId,
337 targetThreadId: sourceThreadId,
338 sourcePoint: { type: "run", runId: firstForkRunId },
339 },
340 {
341 type: "message.dispatch",
342 createdBy: "user",
343 creationSource: "web",
344 commandId: yield* commandId("consume-first"),
345 threadId: sourceThreadId,
346 messageId: MessageId.make(`message-${fixtureName}-consume-first`),
347 text: FIRST_SIBLING_MERGE_USER_TEXT,
348 attachments: [],
349 modelSelection: variant.modelSelection,
350 dispatchMode: { type: "start_immediately" },
351 },
352 {
353 type: "thread.merge_back",
354 createdBy: "user",
355 creationSource: "web",
356 commandId: yield* commandId("merge-second"),
357 sourceThreadId: secondForkThreadId,
358 targetThreadId: sourceThreadId,
359 sourcePoint: { type: "run", runId: secondForkRunId },
360 },
361 {
362 type: "message.dispatch",
363 createdBy: "user",
364 creationSource: "web",
365 commandId: yield* commandId("consume-second"),
366 threadId: sourceThreadId,
367 messageId: MessageId.make(`message-${fixtureName}-consume-second`),
368 text: SECOND_SIBLING_MERGE_USER_TEXT,
369 attachments: [],
370 modelSelection: variant.modelSelection,
371 dispatchMode: { type: "start_immediately" },
372 },
373 {
374 type: "message.dispatch",
375 createdBy: "user",
376 creationSource: "web",
377 commandId: yield* commandId("recall"),
378 threadId: sourceThreadId,
379 messageId: MessageId.make(`message-${fixtureName}-recall`),
380 text: THREAD_MERGE_BACK_SIBLINGS_RECALL_PROMPT,
381 attachments: [],
382 modelSelection: variant.modelSelection,
383 dispatchMode: { type: "start_immediately" },
384 },
385 ] satisfies ReadonlyArray<OrchestrationV2Command>;
386 return {
387 commands,
388 sourceThreadId,
389 firstForkThreadId,
390 secondForkThreadId,
391 firstForkRunId,
392 secondForkRunId,
393 };
394 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
395 const firstSummary = forkDeltaSummary({
396 sourceThreadId: materialized.firstForkThreadId,
397 targetThreadId: materialized.sourceThreadId,
398 forkUserText: THREAD_MERGE_BACK_SIBLINGS_FIRST_FORK_PROMPT,
399 forkAssistantText: "first merge sibling stored",
400 });
401 const secondSummary = forkDeltaSummary({
402 sourceThreadId: materialized.secondForkThreadId,
403 targetThreadId: materialized.sourceThreadId,
404 forkUserText: THREAD_MERGE_BACK_SIBLINGS_SECOND_FORK_PROMPT,
405 forkAssistantText: "second merge sibling stored",
406 });
407 const transcript = parameterizeHandoffs(rawTranscript, [
408 [
409 THREAD_MERGE_BACK_SIBLINGS_FIRST_HANDOFF_PROMPT,
410 variant.driver === "codex"
411 ? FIRST_SIBLING_MERGE_USER_TEXT
412 : providerMessage(firstSummary, FIRST_SIBLING_MERGE_USER_TEXT),
413 ],
414 [
415 THREAD_MERGE_BACK_SIBLINGS_SECOND_HANDOFF_PROMPT,
416 variant.driver === "codex"
417 ? SECOND_SIBLING_MERGE_USER_TEXT
418 : providerMessage(secondSummary, SECOND_SIBLING_MERGE_USER_TEXT),
419 ],
420 ]);
421 const cwd = yield* Effect.acquireRelease(
422 Effect.promise(() => makeCheckpointWorkspace(`merge-back-siblings-${variant.driver}`)),
423 (directory) =>
424 Effect.service(FileSystem.FileSystem).pipe(
425 Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })),
426 Effect.orDie,
427 ),
428 );
429 const steps = [
430 { type: "dispatch" as const, command: materialized.commands[0]!, await: true },
431 { type: "advance_clock" as const, duration: "1 millis" as const },
432 { type: "dispatch" as const, command: materialized.commands[1]!, await: true },
433 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
434 { type: "dispatch" as const, command: materialized.commands[2]!, await: true },
435 { type: "dispatch" as const, command: materialized.commands[3]!, await: true },
436 { type: "await_thread_idle" as const, threadId: materialized.firstForkThreadId },
437 { type: "dispatch" as const, command: materialized.commands[4]!, await: true },
438 { type: "dispatch" as const, command: materialized.commands[5]!, await: true },
439 { type: "await_thread_idle" as const, threadId: materialized.secondForkThreadId },
440 { type: "dispatch" as const, command: materialized.commands[6]!, await: true },
441 { type: "dispatch" as const, command: materialized.commands[7]!, await: true },
442 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
443 { type: "dispatch" as const, command: materialized.commands[8]!, await: true },
444 { type: "dispatch" as const, command: materialized.commands[9]!, await: true },
445 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
446 { type: "dispatch" as const, command: materialized.commands[10]!, await: true },
447 { type: "await_thread_idle" as const, threadId: materialized.sourceThreadId },
448 ];
449 const scenario = {
450 name: `thread_merge_back_siblings/${variant.driver}`,
451 commands: materialized.commands,
452 steps,
453 projectionThreadIds: [
454 materialized.sourceThreadId,
455 materialized.firstForkThreadId,
456 materialized.secondForkThreadId,
457 ],
458 runtimePolicyOverride: { cwd },
459 };
460 const result =
461 variant.driver === "codex"
462 ? yield* runOrchestratorV2ProviderReplayScenario(
463 {
464 ...scenario,
465 transcript: yield* CodexOrchestratorReplayHarness.decodeTranscript(
466 materializeReplayTranscriptWorkspace(transcript, cwd),
467 ),
468 },
469 CodexHistoryReplayHarness,
470 ).pipe(provideDeterministicTestRuntime)
471 : yield* runOrchestratorV2ProviderReplayScenario(
472 {
473 ...scenario,
474 transcript: yield* ClaudeOrchestratorReplayHarness.decodeTranscript(transcript),
475 },
476 ClaudeOrchestratorReplayHarness,
477 ).pipe(provideDeterministicTestRuntime);
478
479 const source = result.projections.get(materialized.sourceThreadId);
480 const firstFork = result.projections.get(materialized.firstForkThreadId);
481 const secondFork = result.projections.get(materialized.secondForkThreadId);
482 assert.isDefined(source);
483 assert.isDefined(firstFork);
484 assert.isDefined(secondFork);
485 assert.equal(
486 source.providerThreads[0]?.nativeThreadRef?.nativeId,
487 nativeSourceThreadId(rawTranscript),
488 );
489 const firstForkNativeId = firstFork.providerThreads[0]?.nativeThreadRef?.nativeId;
490 const secondForkNativeId = secondFork.providerThreads[0]?.nativeThreadRef?.nativeId;
491 assert.isDefined(firstForkNativeId);
492 assert.isDefined(secondForkNativeId);
493 assert.notEqual(firstForkNativeId, secondForkNativeId);
494 assert.notEqual(firstForkNativeId, source.providerThreads[0]?.nativeThreadRef?.nativeId);
495 assert.notEqual(secondForkNativeId, source.providerThreads[0]?.nativeThreadRef?.nativeId);
496 assert.deepEqual(
497 source.contextHandoffs.map((handoff) => handoff.summaryText),
498 [firstSummary, secondSummary],
499 );
500 const mergeTransfers = source.contextTransfers.filter(
501 (transfer) => transfer.type === "merge_back",
502 );
503 assert.lengthOf(mergeTransfers, 2);
504 assert.deepEqual(
505 mergeTransfers.map((transfer) => transfer.status),
506 ["consumed", "consumed"],
507 );
508 assert.deepEqual(
509 mergeTransfers.map((transfer) => transfer.sourcePoint.runId),
510 [materialized.firstForkRunId, materialized.secondForkRunId],
511 );
512 assert.include(
513 normalizePipeSpacing(visibleConversationText(source)),
514 THREAD_MERGE_BACK_SIBLINGS_RECALL,
515 );
516 assert.notInclude(visibleConversationText(source), "Context handoff (");
517 assert.include(visibleConversationText(firstFork), "first merge sibling stored");
518 assert.notInclude(visibleConversationText(firstFork), "second merge sibling stored");
519 assert.include(visibleConversationText(secondFork), "second merge sibling stored");
520 assert.notInclude(visibleConversationText(secondFork), "first merge sibling stored");
521 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
522 );
523});