import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, describe, it } from "@effect/vitest";import { CommandId, MessageId, type OrchestrationV2Command, type OrchestrationV2ThreadProjection, ProviderInstanceId, type ProviderReplayEntry, type ProviderReplayTranscript, ThreadId,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as FileSystem from "effect/FileSystem";import * as Path from "effect/Path";import * as PlatformError from "effect/PlatformError";import * as Schema from "effect/Schema";import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts";import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts";import { IdAllocatorV2, layer as idAllocatorLayer } from "../IdAllocator.ts";import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";import { THREAD_FORK_NATIVE_PRIOR_TURN_ALPHA_PROMPT, THREAD_FORK_NATIVE_PRIOR_TURN_BETA_PROMPT, THREAD_FORK_NATIVE_PRIOR_TURN_REPEAT_PROMPT, THREAD_FORK_NATIVE_SOURCE_PROMPT, THREAD_FORK_NATIVE_TARGET_PROMPT,} from "./fixtures/shared.ts";import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts";import { decodeProviderReplayNdjson, materializeReplayTranscriptWorkspace,} from "./ReplayTranscriptNdjson.ts"; const CODEX_MODEL_SELECTION = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4",} as const;const CLAUDE_MODEL_SELECTION = { instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-sonnet-4-6",} as const;const TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native/codex_transcript.ndjson`;const PRIOR_TURN_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_prior_turn/codex_transcript.ndjson`;const CLAUDE_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native/claude_transcript.ndjson`;const CLAUDE_PRIOR_TURN_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_prior_turn/claude_transcript.ndjson`;const CLAUDE_FORK_LOCAL_ROLLBACK_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_fork_local_rollback/claude_transcript.ndjson`;const CODEX_READ_ONLY_NEVER_POLICY = { approvalPolicy: "never", sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false, },} as const; class ThreadForkGitCommandError extends Schema.TaggedErrorClass<ThreadForkGitCommandError>()( "ThreadForkGitCommandError", { command: Schema.String, exitCode: Schema.Number, },) { override get message(): string { return `${this.command} failed with exit ${this.exitCode}.`; }} function runGit( cwd: string, args: ReadonlyArray<string>,): Effect.Effect< void, ThreadForkGitCommandError | PlatformError.PlatformError, ChildProcessSpawner.ChildProcessSpawner> { return Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const exitCode = yield* spawner.exitCode(ChildProcess.make("git", args, { cwd })); if (Number(exitCode) !== 0) { return yield* new ThreadForkGitCommandError({ command: `git ${args.join(" ")}`, exitCode: Number(exitCode), }); } });} const makeCheckpointWorkspace = Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const cwd = yield* fs.makeTempDirectory({ prefix: "t3-orchestrator-v2-thread-fork-" }); yield* runGit(cwd, ["init"]); yield* runGit(cwd, ["config", "user.name", "T3 Code Test"]); yield* runGit(cwd, ["config", "user.email", "t3code-test@example.com"]); yield* fs.writeFileString(path.join(cwd, "README.md"), "# thread fork\n"); yield* runGit(cwd, ["add", "README.md"]); yield* runGit(cwd, ["commit", "-m", "initial"]); return cwd;}); function readTranscript(transcriptPath: string = TRANSCRIPT_PATH) { return Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const text = yield* fs.readFileString(transcriptPath); return yield* decodeProviderReplayNdjson(text); });} function metadataString(transcript: ProviderReplayTranscript, key: string): string { const value = transcript.metadata?.[key]; if (typeof value !== "string") { throw new Error(`Transcript ${transcript.scenario} is missing metadata string ${key}.`); } return value;} function metadataStringArray( transcript: ProviderReplayTranscript, key: string,): ReadonlyArray<string> { const value = transcript.metadata?.[key]; if (!Array.isArray(value) || !value.every((entry) => typeof entry === "string")) { throw new Error(`Transcript ${transcript.scenario} is missing metadata string array ${key}.`); } return value;} function userAndAssistantText( projection: Pick<OrchestrationV2ThreadProjection, "visibleTurnItems">,): string { return projection.visibleTurnItems .flatMap((row) => { const item = row.item; return item.type === "user_message" || item.type === "assistant_message" ? [item.text] : []; }) .join("\n");} function compactExpectedText(text: string, maxLength = 240): string { const compacted = text.replace(/\s+/g, " ").trim(); if (compacted.length <= maxLength) { return compacted; } return `${compacted.slice(0, maxLength - 3)}...`;} function findCompletedAgentMessageText(input: { readonly transcript: ProviderReplayTranscript; readonly threadId: string; readonly turnId: string;}): string { for (const entry of input.transcript.entries) { if (entry.type !== "emit_inbound") { continue; } const frame = entry.frame as { readonly method?: unknown; readonly params?: { readonly threadId?: unknown; readonly turnId?: unknown; readonly item?: { readonly type?: unknown; readonly text?: unknown; }; }; }; if ( frame.method === "item/completed" && frame.params?.threadId === input.threadId && frame.params.turnId === input.turnId && frame.params.item?.type === "agentMessage" && typeof frame.params.item.text === "string" ) { return frame.params.item.text; } } throw new Error(`No completed agent message found for ${input.threadId}/${input.turnId}`);} function makeExpectedForkDeltaSummary(input: { readonly sourceThreadId: string; readonly targetThreadId: string; readonly forkUserText: string; readonly forkAssistantText: string;}): string { return [ "Merge-back context from forked conversation.", `Source thread: ${input.sourceThreadId}`, `Target thread: ${input.targetThreadId}`, "Covered fork runs: 1-1", "", "Fork delta:", `- User: ${compactExpectedText(input.forkUserText)}`, `- Assistant: ${compactExpectedText(input.forkAssistantText)}`, "- Checkpoint: 0 files", ].join("\n");} function transcriptWithMergeBackContinuation(input: { readonly transcript: ProviderReplayTranscript; readonly providerMessageText: string; readonly projectedUserText: string; readonly assistantText: string;}): ProviderReplayTranscript { const sourceNativeThreadId = "019dd6ba-2681-7bf0-b051-141b0cbcbb27"; const mergeBackNativeTurnId = "019dd6ba-5000-7000-8000-000000000001"; const mergeBackUserItemId = "merge-back-user-message"; const mergeBackAgentItemId = "merge-back-agent-message"; const entriesWithoutExit = input.transcript.entries.filter( (entry) => entry.type !== "runtime_exit", ); const continuation = [ { type: "expect_outbound", label: "turn/start/merge-back-source", frame: { id: 9, method: "turn/start", params: { threadId: sourceNativeThreadId, input: [{ type: "text", text: input.providerMessageText }], approvalPolicy: "never", sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false, }, }, }, }, { type: "emit_inbound", label: "turn/start/merge-back-source", frame: { id: 9, result: { turn: { id: mergeBackNativeTurnId, items: [], status: "inProgress", error: null, startedAt: 1777424041, completedAt: null, durationMs: null, }, }, }, }, { type: "emit_inbound", label: "thread/status/changed/merge-back-source", frame: { method: "thread/status/changed", params: { threadId: sourceNativeThreadId, status: { type: "active", activeFlags: [] }, }, }, }, { type: "emit_inbound", label: "turn/started/merge-back-source", frame: { method: "turn/started", params: { threadId: sourceNativeThreadId, turn: { id: mergeBackNativeTurnId, items: [], status: "inProgress", error: null, startedAt: 1777424041, completedAt: null, durationMs: null, }, }, }, }, { type: "emit_inbound", label: "item/userMessage/started/merge-back-source", frame: { method: "item/started", params: { item: { type: "userMessage", id: mergeBackUserItemId, content: [{ type: "text", text: input.projectedUserText, text_elements: [] }], }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "item/userMessage/completed/merge-back-source", frame: { method: "item/completed", params: { item: { type: "userMessage", id: mergeBackUserItemId, content: [{ type: "text", text: input.projectedUserText, text_elements: [] }], }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "item/agentMessage/started/merge-back-source", frame: { method: "item/started", params: { item: { type: "agentMessage", id: mergeBackAgentItemId, text: "", phase: "final_answer", memoryCitation: null, }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "item/agentMessage/delta/merge-back-source", frame: { method: "item/agentMessage/delta", params: { threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, itemId: mergeBackAgentItemId, delta: input.assistantText, }, }, }, { type: "emit_inbound", label: "item/agentMessage/completed/merge-back-source", frame: { method: "item/completed", params: { item: { type: "agentMessage", id: mergeBackAgentItemId, text: input.assistantText, phase: "final_answer", memoryCitation: null, }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "thread/status/changed/merge-back-source", frame: { method: "thread/status/changed", params: { threadId: sourceNativeThreadId, status: { type: "idle" } }, }, }, { type: "emit_inbound", label: "turn/completed/merge-back-source", frame: { method: "turn/completed", params: { threadId: sourceNativeThreadId, turn: { id: mergeBackNativeTurnId, items: [], status: "completed", error: null, startedAt: 1777424041, completedAt: 1777424042, durationMs: 1000, }, }, }, }, { type: "runtime_exit", status: "success" }, ] satisfies ReadonlyArray<ProviderReplayEntry>; return { ...input.transcript, scenario: `${input.transcript.scenario}_merge_back`, entries: [...entriesWithoutExit, ...continuation], };} describe("orchestration V2 thread fork", () => { it.effect( "creates an idle app fork and resolves it with Codex native thread/fork on first dispatch", () => Effect.gen(function* () { const rawTranscript = yield* readTranscript(); const cwd = yield* Effect.acquireRelease(makeCheckpointWorkspace, (directory) => Effect.service(FileSystem.FileSystem).pipe( Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })), Effect.orDie, ), ); const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript( materializeReplayTranscriptWorkspace(rawTranscript, cwd), ); const materialized = yield* Effect.gen(function* () { const ids = yield* IdAllocatorV2; const projectId = yield* ids.allocate.project({ fixtureName: "thread-fork-native" }); const sourceThreadId = yield* ids.allocate.thread({ fixtureName: "thread-fork-native-source", projectId, }); const targetThreadId = ThreadId.make("thread-fork-native-target"); const commands = [ { type: "thread.create", createdBy: "user", creationSource: "web", commandId: yield* ids.allocate.command({ fixtureName: "thread-fork-native", commandName: "thread-create-source", }), threadId: sourceThreadId, projectId, title: "Source thread", modelSelection: CODEX_MODEL_SELECTION, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, }, { type: "message.dispatch", createdBy: "user", creationSource: "web", commandId: yield* ids.allocate.command({ fixtureName: "thread-fork-native", commandName: "source-message", }), threadId: sourceThreadId, messageId: MessageId.make("message-thread-fork-native-source"), text: THREAD_FORK_NATIVE_SOURCE_PROMPT, attachments: [], modelSelection: CODEX_MODEL_SELECTION, dispatchMode: { type: "start_immediately" }, }, { type: "thread.fork", createdBy: "user", creationSource: "web", commandId: CommandId.make("command-thread-fork-native"), sourceThreadId, targetThreadId, sourcePoint: { type: "latest_stable" }, title: "Forked thread", }, { type: "thread.fork", createdBy: "user", creationSource: "web", commandId: CommandId.make("command-thread-fork-native"), sourceThreadId, targetThreadId, sourcePoint: { type: "latest_stable" }, title: "Forked thread", }, { type: "message.dispatch", createdBy: "user", creationSource: "web", commandId: yield* ids.allocate.command({ fixtureName: "thread-fork-native", commandName: "target-message", }), threadId: targetThreadId, messageId: MessageId.make("message-thread-fork-native-target"), text: THREAD_FORK_NATIVE_TARGET_PROMPT, attachments: [], modelSelection: CODEX_MODEL_SELECTION, dispatchMode: { type: "start_immediately" }, }, ] satisfies ReadonlyArray<OrchestrationV2Command>; return { sourceThreadId, targetThreadId, commands, }; }).pipe(Effect.provide(idAllocatorLayer), provideDeterministicTestRuntime);Lines 1–500 of 1774. Load more
import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, it } from "@effect/vitest"; import { CommandId, MessageId, type OrchestrationV2Command, type OrchestrationV2ThreadProjection, ProviderInstanceId, type ProviderReplayEntry, type ProviderReplayTranscript, ThreadId, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as PlatformError from "effect/PlatformError"; import * as Schema from "effect/Schema"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts"; import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts"; import { IdAllocatorV2, layer as idAllocatorLayer } from "../IdAllocator.ts"; import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts"; import { THREAD_FORK_NATIVE_PRIOR_TURN_ALPHA_PROMPT, THREAD_FORK_NATIVE_PRIOR_TURN_BETA_PROMPT, THREAD_FORK_NATIVE_PRIOR_TURN_REPEAT_PROMPT, THREAD_FORK_NATIVE_SOURCE_PROMPT, THREAD_FORK_NATIVE_TARGET_PROMPT, } from "./fixtures/shared.ts"; import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts"; import { decodeProviderReplayNdjson, materializeReplayTranscriptWorkspace, } from "./ReplayTranscriptNdjson.ts"; const CODEX_MODEL_SELECTION = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4", } as const; const CLAUDE_MODEL_SELECTION = { instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-sonnet-4-6", } as const; const TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native/codex_transcript.ndjson`; const PRIOR_TURN_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_prior_turn/codex_transcript.ndjson`; const CLAUDE_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native/claude_transcript.ndjson`; const CLAUDE_PRIOR_TURN_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_prior_turn/claude_transcript.ndjson`; const CLAUDE_FORK_LOCAL_ROLLBACK_TRANSCRIPT_PATH = `${import.meta.dirname}/fixtures/thread_fork_native_fork_local_rollback/claude_transcript.ndjson`; const CODEX_READ_ONLY_NEVER_POLICY = { approvalPolicy: "never", sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false, }, } as const; class ThreadForkGitCommandError extends Schema.TaggedErrorClass<ThreadForkGitCommandError>()( "ThreadForkGitCommandError", { command: Schema.String, exitCode: Schema.Number, }, ) { override get message(): string { return `${this.command} failed with exit ${this.exitCode}.`; } } function runGit( cwd: string, args: ReadonlyArray<string>, ): Effect.Effect< void, ThreadForkGitCommandError | PlatformError.PlatformError, ChildProcessSpawner.ChildProcessSpawner > { return Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const exitCode = yield* spawner.exitCode(ChildProcess.make("git", args, { cwd })); if (Number(exitCode) !== 0) { return yield* new ThreadForkGitCommandError({ command: `git ${args.join(" ")}`, exitCode: Number(exitCode), }); } }); } const makeCheckpointWorkspace = Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const cwd = yield* fs.makeTempDirectory({ prefix: "t3-orchestrator-v2-thread-fork-" }); yield* runGit(cwd, ["init"]); yield* runGit(cwd, ["config", "user.name", "T3 Code Test"]); yield* runGit(cwd, ["config", "user.email", "t3code-test@example.com"]); yield* fs.writeFileString(path.join(cwd, "README.md"), "# thread fork\n"); yield* runGit(cwd, ["add", "README.md"]); yield* runGit(cwd, ["commit", "-m", "initial"]); return cwd; }); function readTranscript(transcriptPath: string = TRANSCRIPT_PATH) { return Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const text = yield* fs.readFileString(transcriptPath); return yield* decodeProviderReplayNdjson(text); }); } function metadataString(transcript: ProviderReplayTranscript, key: string): string { const value = transcript.metadata?.[key]; if (typeof value !== "string") { throw new Error(`Transcript ${transcript.scenario} is missing metadata string ${key}.`); } return value; } function metadataStringArray( transcript: ProviderReplayTranscript, key: string, ): ReadonlyArray<string> { const value = transcript.metadata?.[key]; if (!Array.isArray(value) || !value.every((entry) => typeof entry === "string")) { throw new Error(`Transcript ${transcript.scenario} is missing metadata string array ${key}.`); } return value; } function userAndAssistantText( projection: Pick<OrchestrationV2ThreadProjection, "visibleTurnItems">, ): string { return projection.visibleTurnItems .flatMap((row) => { const item = row.item; return item.type === "user_message" || item.type === "assistant_message" ? [item.text] : []; }) .join("\n"); } function compactExpectedText(text: string, maxLength = 240): string { const compacted = text.replace(/\s+/g, " ").trim(); if (compacted.length <= maxLength) { return compacted; } return `${compacted.slice(0, maxLength - 3)}...`; } function findCompletedAgentMessageText(input: { readonly transcript: ProviderReplayTranscript; readonly threadId: string; readonly turnId: string; }): string { for (const entry of input.transcript.entries) { if (entry.type !== "emit_inbound") { continue; } const frame = entry.frame as { readonly method?: unknown; readonly params?: { readonly threadId?: unknown; readonly turnId?: unknown; readonly item?: { readonly type?: unknown; readonly text?: unknown; }; }; }; if ( frame.method === "item/completed" && frame.params?.threadId === input.threadId && frame.params.turnId === input.turnId && frame.params.item?.type === "agentMessage" && typeof frame.params.item.text === "string" ) { return frame.params.item.text; } } throw new Error(`No completed agent message found for ${input.threadId}/${input.turnId}`); } function makeExpectedForkDeltaSummary(input: { readonly sourceThreadId: string; readonly targetThreadId: string; readonly forkUserText: string; readonly forkAssistantText: string; }): string { return [ "Merge-back context from forked conversation.", `Source thread: ${input.sourceThreadId}`, `Target thread: ${input.targetThreadId}`, "Covered fork runs: 1-1", "", "Fork delta:", `- User: ${compactExpectedText(input.forkUserText)}`, `- Assistant: ${compactExpectedText(input.forkAssistantText)}`, "- Checkpoint: 0 files", ].join("\n"); } function transcriptWithMergeBackContinuation(input: { readonly transcript: ProviderReplayTranscript; readonly providerMessageText: string; readonly projectedUserText: string; readonly assistantText: string; }): ProviderReplayTranscript { const sourceNativeThreadId = "019dd6ba-2681-7bf0-b051-141b0cbcbb27"; const mergeBackNativeTurnId = "019dd6ba-5000-7000-8000-000000000001"; const mergeBackUserItemId = "merge-back-user-message"; const mergeBackAgentItemId = "merge-back-agent-message"; const entriesWithoutExit = input.transcript.entries.filter( (entry) => entry.type !== "runtime_exit", ); const continuation = [ { type: "expect_outbound", label: "turn/start/merge-back-source", frame: { id: 9, method: "turn/start", params: { threadId: sourceNativeThreadId, input: [{ type: "text", text: input.providerMessageText }], approvalPolicy: "never", sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false, }, }, }, }, { type: "emit_inbound", label: "turn/start/merge-back-source", frame: { id: 9, result: { turn: { id: mergeBackNativeTurnId, items: [], status: "inProgress", error: null, startedAt: 1777424041, completedAt: null, durationMs: null, }, }, }, }, { type: "emit_inbound", label: "thread/status/changed/merge-back-source", frame: { method: "thread/status/changed", params: { threadId: sourceNativeThreadId, status: { type: "active", activeFlags: [] }, }, }, }, { type: "emit_inbound", label: "turn/started/merge-back-source", frame: { method: "turn/started", params: { threadId: sourceNativeThreadId, turn: { id: mergeBackNativeTurnId, items: [], status: "inProgress", error: null, startedAt: 1777424041, completedAt: null, durationMs: null, }, }, }, }, { type: "emit_inbound", label: "item/userMessage/started/merge-back-source", frame: { method: "item/started", params: { item: { type: "userMessage", id: mergeBackUserItemId, content: [{ type: "text", text: input.projectedUserText, text_elements: [] }], }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "item/userMessage/completed/merge-back-source", frame: { method: "item/completed", params: { item: { type: "userMessage", id: mergeBackUserItemId, content: [{ type: "text", text: input.projectedUserText, text_elements: [] }], }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "item/agentMessage/started/merge-back-source", frame: { method: "item/started", params: { item: { type: "agentMessage", id: mergeBackAgentItemId, text: "", phase: "final_answer", memoryCitation: null, }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "item/agentMessage/delta/merge-back-source", frame: { method: "item/agentMessage/delta", params: { threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, itemId: mergeBackAgentItemId, delta: input.assistantText, }, }, }, { type: "emit_inbound", label: "item/agentMessage/completed/merge-back-source", frame: { method: "item/completed", params: { item: { type: "agentMessage", id: mergeBackAgentItemId, text: input.assistantText, phase: "final_answer", memoryCitation: null, }, threadId: sourceNativeThreadId, turnId: mergeBackNativeTurnId, }, }, }, { type: "emit_inbound", label: "thread/status/changed/merge-back-source", frame: { method: "thread/status/changed", params: { threadId: sourceNativeThreadId, status: { type: "idle" } }, }, }, { type: "emit_inbound", label: "turn/completed/merge-back-source", frame: { method: "turn/completed", params: { threadId: sourceNativeThreadId, turn: { id: mergeBackNativeTurnId, items: [], status: "completed", error: null, startedAt: 1777424041, completedAt: 1777424042, durationMs: 1000, }, }, }, }, { type: "runtime_exit", status: "success" }, ] satisfies ReadonlyArray<ProviderReplayEntry>; return { ...input.transcript, scenario: `${input.transcript.scenario}_merge_back`, entries: [...entriesWithoutExit, ...continuation], }; } describe("orchestration V2 thread fork", () => { it.effect( "creates an idle app fork and resolves it with Codex native thread/fork on first dispatch", () => Effect.gen(function* () { const rawTranscript = yield* readTranscript(); const cwd = yield* Effect.acquireRelease(makeCheckpointWorkspace, (directory) => Effect.service(FileSystem.FileSystem).pipe( Effect.flatMap((fs) => fs.remove(directory, { recursive: true, force: true })), Effect.orDie, ), ); const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript( materializeReplayTranscriptWorkspace(rawTranscript, cwd), ); const materialized = yield* Effect.gen(function* () { const ids = yield* IdAllocatorV2; const projectId = yield* ids.allocate.project({ fixtureName: "thread-fork-native" }); const sourceThreadId = yield* ids.allocate.thread({ fixtureName: "thread-fork-native-source", projectId, }); const targetThreadId = ThreadId.make("thread-fork-native-target"); const commands = [ { type: "thread.create", createdBy: "user", creationSource: "web", commandId: yield* ids.allocate.command({ fixtureName: "thread-fork-native", commandName: "thread-create-source", }), threadId: sourceThreadId, projectId, title: "Source thread", modelSelection: CODEX_MODEL_SELECTION, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, }, { type: "message.dispatch", createdBy: "user", creationSource: "web", commandId: yield* ids.allocate.command({ fixtureName: "thread-fork-native", commandName: "source-message", }), threadId: sourceThreadId, messageId: MessageId.make("message-thread-fork-native-source"), text: THREAD_FORK_NATIVE_SOURCE_PROMPT, attachments: [], modelSelection: CODEX_MODEL_SELECTION, dispatchMode: { type: "start_immediately" }, }, { type: "thread.fork", createdBy: "user", creationSource: "web", commandId: CommandId.make("command-thread-fork-native"), sourceThreadId, targetThreadId, sourcePoint: { type: "latest_stable" }, title: "Forked thread", }, { type: "thread.fork", createdBy: "user", creationSource: "web", commandId: CommandId.make("command-thread-fork-native"), sourceThreadId, targetThreadId, sourcePoint: { type: "latest_stable" }, title: "Forked thread", }, { type: "message.dispatch", createdBy: "user", creationSource: "web", commandId: yield* ids.allocate.command({ fixtureName: "thread-fork-native", commandName: "target-message", }), threadId: targetThreadId, messageId: MessageId.make("message-thread-fork-native-target"), text: THREAD_FORK_NATIVE_TARGET_PROMPT, attachments: [], modelSelection: CODEX_MODEL_SELECTION, dispatchMode: { type: "start_immediately" }, }, ] satisfies ReadonlyArray<OrchestrationV2Command>; return { sourceThreadId, targetThreadId, commands, }; }).pipe(Effect.provide(idAllocatorLayer), provideDeterministicTestRuntime);