import * as NodeServices from "@effect/platform-node/NodeServices";import { describe, expect, it, vi } from "@effect/vitest";import { AgentSessionImportProjectChangedError, CommandId, MessageId, ProjectId, ProviderDriverKind, ProviderInstanceId, ThreadId, type OrchestrationCommand, type OrchestrationProjectShell, type OrchestrationThread, type ProviderSendTurnInput,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as Deferred from "effect/Deferred";import * as Fiber from "effect/Fiber";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Path from "effect/Path";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as TestClock from "effect/testing/TestClock"; import { makeTestProviderAdapterHarness } from "../../integration/TestProviderAdapter.integration.ts";import { ServerConfig } from "../config.ts";import { GitWorkflowService } from "../git/GitWorkflowService.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../persistence/Layers/OrchestrationCommandReceipts.ts";import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts";import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts";import { OrchestrationEngineLive } from "../orchestration/Layers/OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "../orchestration/Layers/ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "../orchestration/Layers/ProjectionSnapshotQuery.ts";import { ProviderCommandReactorLive } from "../orchestration/Layers/ProviderCommandReactor.ts";import { OrchestrationCommandInvariantError } from "../orchestration/Errors.ts";import * as ThreadBackgroundLiveness from "../orchestration/ThreadBackgroundLiveness.ts";import * as ThreadPlanProgress from "../orchestration/ThreadPlanProgress.ts";import * as OrchestrationEngine from "../orchestration/Services/OrchestrationEngine.ts";import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts";import { ProviderCommandReactor } from "../orchestration/Services/ProviderCommandReactor.ts";import { ProviderSessionDirectoryLive } from "../provider/Layers/ProviderSessionDirectory.ts";import { makeProviderServiceLive } from "../provider/Layers/ProviderService.ts";import { NoOpProviderEventLoggers, ProviderEventLoggers,} from "../provider/Layers/ProviderEventLoggers.ts";import { ProviderSessionDirectoryPersistenceError } from "../provider/Errors.ts";import { ProviderAdapterRegistry } from "../provider/Services/ProviderAdapterRegistry.ts";import { ProviderAuthService } from "../provider/Services/ProviderAuthService.ts";import * as ProviderSessionDirectory from "../provider/Services/ProviderSessionDirectory.ts";import { makeAdapterRegistryMock } from "../provider/testUtils/providerAdapterRegistryMock.ts";import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts";import { ServerSettingsService } from "../serverSettings.ts";import * as AnalyticsService from "../telemetry/AnalyticsService.ts";import { TextGeneration } from "../textGeneration/TextGeneration.ts";import { VcsStatusBroadcaster } from "../vcs/VcsStatusBroadcaster.ts";import * as RepositoryIdentityResolver from "./RepositoryIdentityResolver.ts";import { importRecentAgentThreads } from "./AgentSessionImporter.ts";import * as AgentSessionScanner from "./AgentSessionScanner.ts"; const PROJECT_ID = ProjectId.make("project-1");const WORKSPACE_ROOT = "/tmp/project-from-server";const CLAUDE_SESSION_ID = "123e4567-e89b-42d3-a456-426614174000";const encodeTranscriptRecord = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const makeThread = (source: "codex" | "claudeAgent"): AgentSessionScanner.AgentSessionThread => ({ source, providerInstanceId: ProviderInstanceId.make(source), providerSessionId: source === "codex" ? "codex-session" : CLAUDE_SESSION_ID, title: `Imported ${source} thread`, model: null, createdAt: "2026-08-24T10:00:00.000Z", updatedAt: "2026-08-24T10:01:00.000Z", messages: [ { role: "user", text: "Fix the bug", createdAt: "2026-08-24T10:00:00.000Z" }, { role: "assistant", text: "Fixed", createdAt: "2026-08-24T10:01:00.000Z" }, ],}); const makeThreadOutcome = (thread: AgentSessionScanner.AgentSessionThread) => ({ _tag: "Importable", thread, source: { provider: thread.source, providerInstanceId: thread.providerInstanceId, providerSessionId: thread.providerSessionId, filePath: `/tmp/transcripts/${thread.providerInstanceId}/${thread.providerSessionId}.jsonl`, size: 0, mtimeMs: 0, device: 0, inode: 0, birthtimeMs: 0, }, }) satisfies AgentSessionScanner.AgentSessionRecentThread; const makeProject = (): OrchestrationProjectShell => ({ id: PROJECT_ID, title: "Project", workspaceRoot: WORKSPACE_ROOT, defaultModelSelection: null, scripts: [], createdAt: "2026-08-24T09:00:00.000Z", updatedAt: "2026-08-24T09:00:00.000Z",}); const makeProjectedThread = (input: { readonly source: "codex" | "claudeAgent"; readonly projectId?: ProjectId; readonly imported?: boolean; readonly includeFollowup?: boolean;}): OrchestrationThread => { const sourceThread = makeThread(input.source); const threadId = ThreadId.make( `import:${sourceThread.providerInstanceId}:${sourceThread.providerSessionId}`, ); return { id: threadId, projectId: input.projectId ?? PROJECT_ID, title: sourceThread.title, modelSelection: { instanceId: sourceThread.providerInstanceId, model: "default" }, runtimeMode: "full-access", interactionMode: "default", pullRequests: [], branch: null, worktreePath: null, latestTurn: null, createdAt: sourceThread.createdAt, updatedAt: sourceThread.updatedAt, archivedAt: null, settledOverride: null, settledAt: null, deletedAt: null, messages: input.imported ? [ { id: MessageId.make(`${threadId}:000000`), role: "user", text: "Fix the bug", turnId: null, streaming: false, createdAt: "2026-08-24T10:00:00.000Z", updatedAt: "2026-08-24T10:00:00.000Z", }, ...(input.includeFollowup ? [ { id: MessageId.make("user-followup"), role: "user" as const, text: "Keep going", turnId: null, streaming: false, createdAt: "2026-08-24T10:02:00.000Z", updatedAt: "2026-08-24T10:02:00.000Z", }, ] : []), ] : [], proposedPlans: [], activities: [], checkpoints: [], session: null, };}; const makeSnapshotsLayer = (input: { readonly project?: OrchestrationProjectShell; readonly getThread?: (threadId: ThreadId) => Option.Option<OrchestrationThread>;}) => Layer.mock(ProjectionSnapshotQuery.ProjectionSnapshotQuery)({ getProjectShellById: () => Effect.succeed(input.project === undefined ? Option.none() : Option.some(input.project)), getImportedAgentSessionSources: () => Effect.succeed([]), getThreadDetailById: (threadId) => Effect.succeed(input.getThread?.(threadId) ?? Option.none()), }); const runImport = (input: { readonly scanner: AgentSessionScanner.AgentSessionScanner["Service"]; readonly engine: OrchestrationEngine.OrchestrationEngineService["Service"]; readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"]; readonly snapshots: ReturnType<typeof makeSnapshotsLayer>; readonly expectedWorkspaceRoot?: string;}) => importRecentAgentThreads({ projectId: PROJECT_ID, ...(input.expectedWorkspaceRoot === undefined ? {} : { expectedWorkspaceRoot: input.expectedWorkspaceRoot }), }).pipe( Effect.provideService(AgentSessionScanner.AgentSessionScanner, input.scanner), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, input.engine), Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, input.directory), Effect.provide(input.snapshots), ); it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { describe("importRecentAgentThreads", () => { it.effect("uses the project root and stores provider-specific resume cursors", () => Effect.gen(function* () { const commands: Array<OrchestrationCommand> = []; const bindings: Array<ProviderSessionDirectory.ProviderRuntimeBinding> = []; let scannedRoot: string | undefined; const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: (workspaceRoot) => { scannedRoot = workspaceRoot; return Stream.concat( Stream.succeed(makeThreadOutcome(makeThread("codex"))), Stream.fromEffect( Effect.sync(() => { expect(bindings).toHaveLength(1); return makeThreadOutcome(makeThread("claudeAgent")); }), ), ); }, }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: (command) => Effect.sync(() => ({ sequence: commands.push(command) })), readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: (binding) => Effect.sync(() => void bindings.push(binding)), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeed(Option.none()), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const result = yield* runImport({ scanner, engine, directory, snapshots: makeSnapshotsLayer({ project: makeProject() }), expectedWorkspaceRoot: `${WORKSPACE_ROOT}/`, }); expect(result).toEqual({ importedCount: 2, skippedCount: 0 }); expect(scannedRoot).toBe(WORKSPACE_ROOT); expect(commands.map((command) => command.type)).toEqual([ "thread.create", "thread.history.import", "thread.create", "thread.history.import", ]); expect(commands.filter((command) => command.type === "thread.create")).toMatchObject([ { historyImport: true }, { historyImport: true }, ]); expect( commands .filter((command) => command.type === "thread.history.import") .flatMap((command) => command.messages.map((message) => message.messageId)), ).toEqual([ "import:codex:codex-session:000000", "import:codex:codex-session:000001", `import:claudeAgent:${CLAUDE_SESSION_ID}:000000`, `import:claudeAgent:${CLAUDE_SESSION_ID}:000001`, ]); expect(bindings).toMatchObject([ { provider: "codex", providerInstanceId: "codex", resumeCursor: { threadId: "codex-session" }, runtimePayload: { cwd: WORKSPACE_ROOT }, }, { provider: "claudeAgent", providerInstanceId: "claudeAgent", resumeCursor: { threadId: `import:claudeAgent:${CLAUDE_SESSION_ID}`, resume: CLAUDE_SESSION_ID, }, runtimePayload: { cwd: WORKSPACE_ROOT }, }, ]); }), ); it.effect("rejects a changed project root before scanning or writing", () => Effect.gen(function* () { const recentThreads = vi.fn(() => Stream.empty); const error = yield* importRecentAgentThreads({ projectId: PROJECT_ID, expectedWorkspaceRoot: WORKSPACE_ROOT, }).pipe( Effect.provideService( AgentSessionScanner.AgentSessionScanner, AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("must not scan a changed project"), recentThreads, }), ), Effect.provide( Layer.mergeAll( Layer.mock(OrchestrationEngine.OrchestrationEngineService)({}), Layer.mock(ProviderSessionDirectory.ProviderSessionDirectory)({}), makeSnapshotsLayer({ project: { ...makeProject(), workspaceRoot: "/tmp/project-moved" }, }), ), ), Effect.flip, ); expect(error).toEqual(new AgentSessionImportProjectChangedError({ projectId: PROJECT_ID })); expect(recentThreads).not.toHaveBeenCalled(); }), ); it.effect("counts scanner skips without writing a thread or binding", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.succeed({ _tag: "Skipped" }), }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: () => Effect.die("must not dispatch for a scanner skip"), readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: () => Effect.die("must not bind a scanner skip"), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.die("unused"), getBinding: () => Effect.die("must not read a scanner skip binding"), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const result = yield* runImport({ scanner, engine, directory, snapshots: makeSnapshotsLayer({ project: makeProject() }), }); expect(result).toEqual({ importedCount: 0, skippedCount: 1 }); }), ); it.effect("recovers after a rejected history receipt and a failed binding write", () => Effect.gen(function* () { let threadCreated = false; let historyImported = false; let historyAttemptCount = 0; let bindingAttemptCount = 0; const rejectedCommandIds = new Set<string>(); const bindings: Array<ProviderSessionDirectory.ProviderRuntimeBinding> = []; const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.fromIterable([makeThreadOutcome(makeThread("codex"))]), }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: (command) => { if (rejectedCommandIds.has(command.commandId)) { return Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: "Previously rejected.", }), ); } if (command.type === "thread.create") threadCreated = true; if (command.type === "thread.history.import") { historyAttemptCount += 1; if (historyAttemptCount === 1) { rejectedCommandIds.add(command.commandId); return Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: "Temporary history import failure.", }), ); } historyImported = true; } return Effect.succeed({ sequence: 1 }); }, readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: (binding) => { bindingAttemptCount += 1; if (bindingAttemptCount === 1) { return Effect.fail( new ProviderSessionDirectoryPersistenceError({ operation: "upsert", detail: "Temporary session storage failure.", }), ); } bindings.push(binding); return Effect.void; }, getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeed(bindings[0] === undefined ? Option.none() : Option.some(bindings[0])), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const snapshots = makeSnapshotsLayer({ project: makeProject(), getThread: () => threadCreated ? Option.some(makeProjectedThread({ source: "codex", imported: historyImported })) : Option.none(), }); const importOnce = () => runImport({ scanner, engine, directory, snapshots }); expect(yield* importOnce()).toEqual({ importedCount: 0, skippedCount: 1 }); expect(yield* importOnce()).toEqual({ importedCount: 0, skippedCount: 1 }); expect(yield* importOnce()).toEqual({ importedCount: 1, skippedCount: 0 }); const historyAttemptsAfterCompletion = historyAttemptCount; expect(yield* importOnce()).toEqual({ importedCount: 1, skippedCount: 0 }); expect(historyAttemptCount).toBe(historyAttemptsAfterCompletion); expect(historyAttemptCount).toBe(2); expect(bindings).toHaveLength(1); }), ); it.effect("does not replace completed history or an active binding on retry", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.fromIterable([makeThreadOutcome(makeThread("codex"))]), }); const runningBinding: ProviderSessionDirectory.ProviderRuntimeBinding = { threadId: ThreadId.make("import:codex:codex-session"), provider: ProviderDriverKind.make("codex"), providerInstanceId: ProviderInstanceId.make("codex"), status: "running", resumeCursor: { threadId: "newer-codex-session" }, }; const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: () => Effect.die("must not replace an active binding"), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeed(Option.some(runningBinding)), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: () => Effect.die("must not replay history or settle active work"), readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const result = yield* runImport({ scanner, engine, directory, snapshots: makeSnapshotsLayer({ project: makeProject(), getThread: () => Option.some( makeProjectedThread({ source: "codex", imported: true, includeFollowup: true }), ), }), }); expect(result).toEqual({ importedCount: 1, skippedCount: 0 }); }), ); it.effect("skips malformed Claude ids and wrong-project thread collisions", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.fromIterable([ makeThreadOutcome({ ...makeThread("claudeAgent"), providerSessionId: "not-a-uuid" }), makeThreadOutcome(makeThread("codex")), ]), }); const commands: Array<OrchestrationCommand> = [];Lines 1–500 of 1214. Load more
import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it, vi } from "@effect/vitest"; import { AgentSessionImportProjectChangedError, CommandId, MessageId, ProjectId, ProviderDriverKind, ProviderInstanceId, ThreadId, type OrchestrationCommand, type OrchestrationProjectShell, type OrchestrationThread, type ProviderSendTurnInput, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { makeTestProviderAdapterHarness } from "../../integration/TestProviderAdapter.integration.ts"; import { ServerConfig } from "../config.ts"; import { GitWorkflowService } from "../git/GitWorkflowService.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../persistence/Layers/OrchestrationCommandReceipts.ts"; import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts"; import { OrchestrationEngineLive } from "../orchestration/Layers/OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "../orchestration/Layers/ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "../orchestration/Layers/ProjectionSnapshotQuery.ts"; import { ProviderCommandReactorLive } from "../orchestration/Layers/ProviderCommandReactor.ts"; import { OrchestrationCommandInvariantError } from "../orchestration/Errors.ts"; import * as ThreadBackgroundLiveness from "../orchestration/ThreadBackgroundLiveness.ts"; import * as ThreadPlanProgress from "../orchestration/ThreadPlanProgress.ts"; import * as OrchestrationEngine from "../orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts"; import { ProviderCommandReactor } from "../orchestration/Services/ProviderCommandReactor.ts"; import { ProviderSessionDirectoryLive } from "../provider/Layers/ProviderSessionDirectory.ts"; import { makeProviderServiceLive } from "../provider/Layers/ProviderService.ts"; import { NoOpProviderEventLoggers, ProviderEventLoggers, } from "../provider/Layers/ProviderEventLoggers.ts"; import { ProviderSessionDirectoryPersistenceError } from "../provider/Errors.ts"; import { ProviderAdapterRegistry } from "../provider/Services/ProviderAdapterRegistry.ts"; import { ProviderAuthService } from "../provider/Services/ProviderAuthService.ts"; import * as ProviderSessionDirectory from "../provider/Services/ProviderSessionDirectory.ts"; import { makeAdapterRegistryMock } from "../provider/testUtils/providerAdapterRegistryMock.ts"; import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts"; import { ServerSettingsService } from "../serverSettings.ts"; import * as AnalyticsService from "../telemetry/AnalyticsService.ts"; import { TextGeneration } from "../textGeneration/TextGeneration.ts"; import { VcsStatusBroadcaster } from "../vcs/VcsStatusBroadcaster.ts"; import * as RepositoryIdentityResolver from "./RepositoryIdentityResolver.ts"; import { importRecentAgentThreads } from "./AgentSessionImporter.ts"; import * as AgentSessionScanner from "./AgentSessionScanner.ts"; const PROJECT_ID = ProjectId.make("project-1"); const WORKSPACE_ROOT = "/tmp/project-from-server"; const CLAUDE_SESSION_ID = "123e4567-e89b-42d3-a456-426614174000"; const encodeTranscriptRecord = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const makeThread = (source: "codex" | "claudeAgent"): AgentSessionScanner.AgentSessionThread => ({ source, providerInstanceId: ProviderInstanceId.make(source), providerSessionId: source === "codex" ? "codex-session" : CLAUDE_SESSION_ID, title: `Imported ${source} thread`, model: null, createdAt: "2026-08-24T10:00:00.000Z", updatedAt: "2026-08-24T10:01:00.000Z", messages: [ { role: "user", text: "Fix the bug", createdAt: "2026-08-24T10:00:00.000Z" }, { role: "assistant", text: "Fixed", createdAt: "2026-08-24T10:01:00.000Z" }, ], }); const makeThreadOutcome = (thread: AgentSessionScanner.AgentSessionThread) => ({ _tag: "Importable", thread, source: { provider: thread.source, providerInstanceId: thread.providerInstanceId, providerSessionId: thread.providerSessionId, filePath: `/tmp/transcripts/${thread.providerInstanceId}/${thread.providerSessionId}.jsonl`, size: 0, mtimeMs: 0, device: 0, inode: 0, birthtimeMs: 0, }, }) satisfies AgentSessionScanner.AgentSessionRecentThread; const makeProject = (): OrchestrationProjectShell => ({ id: PROJECT_ID, title: "Project", workspaceRoot: WORKSPACE_ROOT, defaultModelSelection: null, scripts: [], createdAt: "2026-08-24T09:00:00.000Z", updatedAt: "2026-08-24T09:00:00.000Z", }); const makeProjectedThread = (input: { readonly source: "codex" | "claudeAgent"; readonly projectId?: ProjectId; readonly imported?: boolean; readonly includeFollowup?: boolean; }): OrchestrationThread => { const sourceThread = makeThread(input.source); const threadId = ThreadId.make( `import:${sourceThread.providerInstanceId}:${sourceThread.providerSessionId}`, ); return { id: threadId, projectId: input.projectId ?? PROJECT_ID, title: sourceThread.title, modelSelection: { instanceId: sourceThread.providerInstanceId, model: "default" }, runtimeMode: "full-access", interactionMode: "default", pullRequests: [], branch: null, worktreePath: null, latestTurn: null, createdAt: sourceThread.createdAt, updatedAt: sourceThread.updatedAt, archivedAt: null, settledOverride: null, settledAt: null, deletedAt: null, messages: input.imported ? [ { id: MessageId.make(`${threadId}:000000`), role: "user", text: "Fix the bug", turnId: null, streaming: false, createdAt: "2026-08-24T10:00:00.000Z", updatedAt: "2026-08-24T10:00:00.000Z", }, ...(input.includeFollowup ? [ { id: MessageId.make("user-followup"), role: "user" as const, text: "Keep going", turnId: null, streaming: false, createdAt: "2026-08-24T10:02:00.000Z", updatedAt: "2026-08-24T10:02:00.000Z", }, ] : []), ] : [], proposedPlans: [], activities: [], checkpoints: [], session: null, }; }; const makeSnapshotsLayer = (input: { readonly project?: OrchestrationProjectShell; readonly getThread?: (threadId: ThreadId) => Option.Option<OrchestrationThread>; }) => Layer.mock(ProjectionSnapshotQuery.ProjectionSnapshotQuery)({ getProjectShellById: () => Effect.succeed(input.project === undefined ? Option.none() : Option.some(input.project)), getImportedAgentSessionSources: () => Effect.succeed([]), getThreadDetailById: (threadId) => Effect.succeed(input.getThread?.(threadId) ?? Option.none()), }); const runImport = (input: { readonly scanner: AgentSessionScanner.AgentSessionScanner["Service"]; readonly engine: OrchestrationEngine.OrchestrationEngineService["Service"]; readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"]; readonly snapshots: ReturnType<typeof makeSnapshotsLayer>; readonly expectedWorkspaceRoot?: string; }) => importRecentAgentThreads({ projectId: PROJECT_ID, ...(input.expectedWorkspaceRoot === undefined ? {} : { expectedWorkspaceRoot: input.expectedWorkspaceRoot }), }).pipe( Effect.provideService(AgentSessionScanner.AgentSessionScanner, input.scanner), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, input.engine), Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, input.directory), Effect.provide(input.snapshots), ); it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { describe("importRecentAgentThreads", () => { it.effect("uses the project root and stores provider-specific resume cursors", () => Effect.gen(function* () { const commands: Array<OrchestrationCommand> = []; const bindings: Array<ProviderSessionDirectory.ProviderRuntimeBinding> = []; let scannedRoot: string | undefined; const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: (workspaceRoot) => { scannedRoot = workspaceRoot; return Stream.concat( Stream.succeed(makeThreadOutcome(makeThread("codex"))), Stream.fromEffect( Effect.sync(() => { expect(bindings).toHaveLength(1); return makeThreadOutcome(makeThread("claudeAgent")); }), ), ); }, }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: (command) => Effect.sync(() => ({ sequence: commands.push(command) })), readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: (binding) => Effect.sync(() => void bindings.push(binding)), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeed(Option.none()), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const result = yield* runImport({ scanner, engine, directory, snapshots: makeSnapshotsLayer({ project: makeProject() }), expectedWorkspaceRoot: `${WORKSPACE_ROOT}/`, }); expect(result).toEqual({ importedCount: 2, skippedCount: 0 }); expect(scannedRoot).toBe(WORKSPACE_ROOT); expect(commands.map((command) => command.type)).toEqual([ "thread.create", "thread.history.import", "thread.create", "thread.history.import", ]); expect(commands.filter((command) => command.type === "thread.create")).toMatchObject([ { historyImport: true }, { historyImport: true }, ]); expect( commands .filter((command) => command.type === "thread.history.import") .flatMap((command) => command.messages.map((message) => message.messageId)), ).toEqual([ "import:codex:codex-session:000000", "import:codex:codex-session:000001", `import:claudeAgent:${CLAUDE_SESSION_ID}:000000`, `import:claudeAgent:${CLAUDE_SESSION_ID}:000001`, ]); expect(bindings).toMatchObject([ { provider: "codex", providerInstanceId: "codex", resumeCursor: { threadId: "codex-session" }, runtimePayload: { cwd: WORKSPACE_ROOT }, }, { provider: "claudeAgent", providerInstanceId: "claudeAgent", resumeCursor: { threadId: `import:claudeAgent:${CLAUDE_SESSION_ID}`, resume: CLAUDE_SESSION_ID, }, runtimePayload: { cwd: WORKSPACE_ROOT }, }, ]); }), ); it.effect("rejects a changed project root before scanning or writing", () => Effect.gen(function* () { const recentThreads = vi.fn(() => Stream.empty); const error = yield* importRecentAgentThreads({ projectId: PROJECT_ID, expectedWorkspaceRoot: WORKSPACE_ROOT, }).pipe( Effect.provideService( AgentSessionScanner.AgentSessionScanner, AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("must not scan a changed project"), recentThreads, }), ), Effect.provide( Layer.mergeAll( Layer.mock(OrchestrationEngine.OrchestrationEngineService)({}), Layer.mock(ProviderSessionDirectory.ProviderSessionDirectory)({}), makeSnapshotsLayer({ project: { ...makeProject(), workspaceRoot: "/tmp/project-moved" }, }), ), ), Effect.flip, ); expect(error).toEqual(new AgentSessionImportProjectChangedError({ projectId: PROJECT_ID })); expect(recentThreads).not.toHaveBeenCalled(); }), ); it.effect("counts scanner skips without writing a thread or binding", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.succeed({ _tag: "Skipped" }), }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: () => Effect.die("must not dispatch for a scanner skip"), readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: () => Effect.die("must not bind a scanner skip"), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.die("unused"), getBinding: () => Effect.die("must not read a scanner skip binding"), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const result = yield* runImport({ scanner, engine, directory, snapshots: makeSnapshotsLayer({ project: makeProject() }), }); expect(result).toEqual({ importedCount: 0, skippedCount: 1 }); }), ); it.effect("recovers after a rejected history receipt and a failed binding write", () => Effect.gen(function* () { let threadCreated = false; let historyImported = false; let historyAttemptCount = 0; let bindingAttemptCount = 0; const rejectedCommandIds = new Set<string>(); const bindings: Array<ProviderSessionDirectory.ProviderRuntimeBinding> = []; const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.fromIterable([makeThreadOutcome(makeThread("codex"))]), }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: (command) => { if (rejectedCommandIds.has(command.commandId)) { return Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: "Previously rejected.", }), ); } if (command.type === "thread.create") threadCreated = true; if (command.type === "thread.history.import") { historyAttemptCount += 1; if (historyAttemptCount === 1) { rejectedCommandIds.add(command.commandId); return Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: "Temporary history import failure.", }), ); } historyImported = true; } return Effect.succeed({ sequence: 1 }); }, readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: (binding) => { bindingAttemptCount += 1; if (bindingAttemptCount === 1) { return Effect.fail( new ProviderSessionDirectoryPersistenceError({ operation: "upsert", detail: "Temporary session storage failure.", }), ); } bindings.push(binding); return Effect.void; }, getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeed(bindings[0] === undefined ? Option.none() : Option.some(bindings[0])), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const snapshots = makeSnapshotsLayer({ project: makeProject(), getThread: () => threadCreated ? Option.some(makeProjectedThread({ source: "codex", imported: historyImported })) : Option.none(), }); const importOnce = () => runImport({ scanner, engine, directory, snapshots }); expect(yield* importOnce()).toEqual({ importedCount: 0, skippedCount: 1 }); expect(yield* importOnce()).toEqual({ importedCount: 0, skippedCount: 1 }); expect(yield* importOnce()).toEqual({ importedCount: 1, skippedCount: 0 }); const historyAttemptsAfterCompletion = historyAttemptCount; expect(yield* importOnce()).toEqual({ importedCount: 1, skippedCount: 0 }); expect(historyAttemptCount).toBe(historyAttemptsAfterCompletion); expect(historyAttemptCount).toBe(2); expect(bindings).toHaveLength(1); }), ); it.effect("does not replace completed history or an active binding on retry", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.fromIterable([makeThreadOutcome(makeThread("codex"))]), }); const runningBinding: ProviderSessionDirectory.ProviderRuntimeBinding = { threadId: ThreadId.make("import:codex:codex-session"), provider: ProviderDriverKind.make("codex"), providerInstanceId: ProviderInstanceId.make("codex"), status: "running", resumeCursor: { threadId: "newer-codex-session" }, }; const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ upsert: () => Effect.die("must not replace an active binding"), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeed(Option.some(runningBinding)), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.die("unused"), }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: () => Effect.die("must not replay history or settle active work"), readEvents: () => Stream.empty, readThreadEvents: () => Stream.empty, getThreadReplayStats: () => Effect.die("unused"), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }); const result = yield* runImport({ scanner, engine, directory, snapshots: makeSnapshotsLayer({ project: makeProject(), getThread: () => Option.some( makeProjectedThread({ source: "codex", imported: true, includeFollowup: true }), ), }), }); expect(result).toEqual({ importedCount: 1, skippedCount: 0 }); }), ); it.effect("skips malformed Claude ids and wrong-project thread collisions", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ scan: Effect.die("unused"), recentThreads: () => Stream.fromIterable([ makeThreadOutcome({ ...makeThread("claudeAgent"), providerSessionId: "not-a-uuid" }), makeThreadOutcome(makeThread("codex")), ]), }); const commands: Array<OrchestrationCommand> = [];