1import { expect, it } from "@effect/vitest";
2import {
3 ProjectId,
4 ProviderInstanceId,
5 ThreadId,
6 type OrchestrationV2DomainEvent,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10import * as Option from "effect/Option";
11import * as Stream from "effect/Stream";
12
13import * as EventSink from "../orchestration-v2/EventSink.ts";
14import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
15import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
16import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts";
17import * as AgentSessionImporter from "./AgentSessionImporter.ts";
18import * as AgentSessionScanner from "./AgentSessionScanner.ts";
19import * as ProjectService from "./ProjectService.ts";
20
21const projectId = ProjectId.make("agent-session-import-project");
22const providerInstanceId = ProviderInstanceId.make("codex");
23const providerSessionId = "native-codex-thread";
24const threadId = ThreadId.make(`import:${providerInstanceId}:${providerSessionId}`);
25
26it.effect("imports messages once and preserves the provider native resume binding", () => {
27 const writes: Array<ReadonlyArray<OrchestrationV2DomainEvent>> = [];
28 const upserts: Array<unknown> = [];
29 const recorded: Array<unknown> = [];
30 let imported = false;
31 const scanner = AgentSessionScanner.AgentSessionScanner.of({
32 scan: Effect.die("unused"),
33 recentThreads: () =>
34 Stream.succeed({
35 _tag: "Importable",
36 source: {
37 provider: "codex",
38 providerInstanceId,
39 providerSessionId,
40 filePath: "/tmp/native-codex-thread.jsonl",
41 size: 100,
42 mtimeMs: 2,
43 device: 3,
44 inode: 4,
45 birthtimeMs: 1,
46 },
47 thread: {
48 source: "codex",
49 providerInstanceId,
50 providerSessionId,
51 title: "Imported thread",
52 model: "gpt-5.4",
53 createdAt: "2026-09-01T10:00:00.000Z",
54 updatedAt: "2026-09-01T10:01:00.000Z",
55 messages: [
56 { role: "user", text: "Fix it", createdAt: "2026-09-01T10:00:00.000Z" },
57 { role: "assistant", text: "Fixed", createdAt: "2026-09-01T10:01:00.000Z" },
58 ],
59 },
60 }),
61 });
62 const layerTest = AgentSessionImporter.layer.pipe(
63 Layer.provide(
64 Layer.mergeAll(
65 Layer.succeed(AgentSessionScanner.AgentSessionScanner, scanner),
66 Layer.mock(ProjectService.ProjectService)({
67 getById: () =>
68 Effect.succeed(
69 Option.some({ id: projectId, workspaceRoot: "/workspace/project" } as never),
70 ),
71 }),
72 Layer.mock(Orchestrator.OrchestratorV2)({
73 getThreadRecords: () =>
74 imported
75 ? Effect.succeed({
76 thread: { id: threadId, projectId, historyOrigin: "v1_import" },
77 } as never)
78 : Effect.fail(new Orchestrator.OrchestratorProjectionError({ threadId })),
79 }),
80 Layer.mock(EventSink.EventSinkV2)({
81 write: (input) =>
82 Effect.sync(() => {
83 writes.push(input.events);
84 imported = true;
85 return [];
86 }),
87 }),
88 Layer.mock(ProviderSessionRuntime.ProviderSessionRuntimeRepository)({
89 list: () => Effect.succeed([]),
90 upsert: (input) => Effect.sync(() => void upserts.push(input)),
91 recordImportedTranscript: (input) => Effect.sync(() => void recorded.push(input)),
92 }),
93 IdAllocator.layer,
94 ),
95 ),
96 );
97
98 return Effect.gen(function* () {
99 const importer = yield* AgentSessionImporter.AgentSessionImporter;
100 expect(yield* importer.importRecentAgentThreads({ projectId })).toEqual({
101 importedCount: 1,
102 skippedCount: 0,
103 });
104 expect(yield* importer.importRecentAgentThreads({ projectId })).toEqual({
105 importedCount: 1,
106 skippedCount: 0,
107 });
108
109 expect(writes).toHaveLength(1);
110 expect(writes[0]?.map((event) => event.type)).toEqual([
111 "thread.created",
112 "message.updated",
113 "turn-item.updated",
114 "message.updated",
115 "turn-item.updated",
116 "provider-thread.updated",
117 ]);
118 const created = writes[0]?.find((event) => event.type === "thread.created");
119 const providerThread = writes[0]?.find((event) => event.type === "provider-thread.updated");
120 expect(created?.payload).toMatchObject({
121 id: threadId,
122 activeProviderThreadId: providerThread?.payload.id,
123 historyOrigin: "v1_import",
124 });
125 expect(providerThread?.payload).toMatchObject({
126 appThreadId: threadId,
127 nativeThreadRef: {
128 driver: "codex",
129 nativeId: providerSessionId,
130 strength: "strong",
131 },
132 });
133 expect(
134 writes[0]
135 ?.filter((event) => event.type === "message.updated")
136 .map((event) => event.payload.text),
137 ).toEqual(["Fix it", "Fixed"]);
138 expect(upserts).toEqual([
139 expect.objectContaining({
140 threadId,
141 providerInstanceId,
142 resumeCursor: { threadId: providerSessionId },
143 }),
144 ]);
145 expect(recorded).toHaveLength(2);
146 }).pipe(Effect.provide(layerTest));
147});