1import { expect, it } from "@effect/vitest";
2import {
3 type AgentSessionSource,
4 type ServerSettingsError,
5 EventId,
6 ProjectId,
7 ProviderDriverKind,
8 ProviderInstanceId,
9 MessageId,
10 RunId,
11 ThreadId,
12 type ApplicationStoredEvent,
13 type OrchestrationV2AppThread,
14 type OrchestrationV2DomainEvent,
15 type OrchestrationV2ProviderThread,
16} from "@t3tools/contracts";
17import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
18import * as KeyedLock from "@t3tools/shared/KeyedLock";
19import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator";
20import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
21import * as Context from "effect/Context";
22import * as DateTime from "effect/DateTime";
23import * as Deferred from "effect/Deferred";
24import * as Effect from "effect/Effect";
25import * as Fiber from "effect/Fiber";
26import * as Layer from "effect/Layer";
27import * as Option from "effect/Option";
28import * as Stream from "effect/Stream";
29import * as SqlClient from "effect/sql/SqlClient";
30import * as TestClock from "effect/testing/TestClock";
31
32import * as EventSink from "../orchestration-v2/EventSink.ts";
33import * as EventStore from "../orchestration-v2/EventStore.ts";
34import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
35import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
36import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
37import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts";
38import {
39 CLAUDE_MODEL_SELECTION,
40 CODEX_MODEL_SELECTION,
41} from "../orchestration-v2/testkit/fixtures/shared.ts";
42import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts";
43import * as SqlitePersistence from "../persistence/Sqlite.ts";
44import * as ServerSettings from "../serverSettings.ts";
45import * as AgentSessionImporter from "./AgentSessionImporter.ts";
46import * as AgentSessionScanner from "./AgentSessionScanner.ts";
47import * as ProjectService from "./ProjectService.ts";
48
49const projectId = ProjectId.make("agent-session-import-project");
50const timestamp = "2026-09-01T10:00:00.000Z";
51const now = DateTime.makeUnsafe(timestamp);
52const stores = Layer.mergeAll(EventStore.layer, ProjectionStore.layer).pipe(
53 Layer.provideMerge(SqlitePersistence.layerMemory),
54);
55const persistence = Layer.mergeAll(
56 stores,
57 EventSink.layer.pipe(Layer.provide(stores)),
58 ProviderSessionRuntime.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
59 IdAllocator.layer,
60);
61
62function candidate(
63 modelSelection: typeof CODEX_MODEL_SELECTION,
64 providerSessionId: string,
65 provider: AgentSessionSource = "codex",
66) {
67 const providerInstanceId = modelSelection.instanceId;
68 return {
69 _tag: "Importable" as const,
70 source: {
71 provider,
72 providerInstanceId,
73 providerSessionId,
74 filePath: `/tmp/${providerInstanceId}/${providerSessionId}.jsonl`,
75 size: 100,
76 mtimeMs: 2,
77 device: 3,
78 inode: 4,
79 birthtimeMs: 1,
80 },
81 thread: {
82 source: provider,
83 providerInstanceId,
84 providerSessionId,
85 title: "Imported thread",
86 model: modelSelection.model,
87 createdAt: timestamp,
88 updatedAt: timestamp,
89 omittedMessageCount: 0,
90 messages: [
91 { role: "user" as const, text: "Fix it", createdAt: timestamp },
92 { role: "assistant" as const, text: "Fixed", createdAt: timestamp },
93 ],
94 },
95 } satisfies AgentSessionScanner.AgentSessionRecentThread;
96}
97
98const PI_MODEL_SELECTION = { instanceId: ProviderInstanceId.make("pi"), model: "default" };
99
100// Pi threads are owned by their session file path, not the session UUID.
101const nativeId = ({ source }: ReturnType<typeof candidate>) =>
102 source.provider === "pi" ? source.filePath : source.providerSessionId;
103
104function importId(source: ReturnType<typeof candidate>) {
105 return ThreadId.make(
106 `import:${source.source.providerInstanceId}:${source.source.providerSessionId}`,
107 );
108}
109
110const seedOwner = Effect.fn(function* (
111 source: ReturnType<typeof candidate>,
112 lifecycle: "active" | "archived" | "deleted" = "active",
113) {
114 const sink = yield* EventSink.EventSinkV2;
115 const ids = yield* IdAllocator.IdAllocatorV2;
116 const { providerInstanceId, providerSessionId, provider } = source.source;
117 const nativeThreadId = nativeId(source);
118 const threadId = ThreadId.make(`owner:${providerInstanceId}:${providerSessionId}`);
119 const driver = ProviderDriverKind.make(provider);
120 const providerThread: OrchestrationV2ProviderThread = {
121 id: ids.derive.providerThread({ driver, nativeThreadId }),
122 driver,
123 providerInstanceId,
124 providerSessionId: null,
125 appThreadId: threadId,
126 ownerNodeId: null,
127 nativeThreadRef: { driver, nativeId: nativeThreadId, strength: "strong" },
128 nativeConversationHeadRef: null,
129 status: "idle",
130 firstRunOrdinal: 1,
131 lastRunOrdinal: 1,
132 handoffIds: [],
133 forkedFrom: null,
134 createdAt: now,
135 updatedAt: now,
136 };
137 const thread: OrchestrationV2AppThread = {
138 createdBy: "user",
139 creationSource: "web",
140 id: threadId,
141 projectId,
142 title: "Native thread",
143 providerInstanceId,
144 modelSelection: { instanceId: providerInstanceId, model: source.thread.model },
145 runtimeMode: "full-access",
146 interactionMode: "default",
147 branch: null,
148 worktreePath: null,
149 activeProviderThreadId: providerThread.id,
150 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
151 forkedFrom: null,
152 createdAt: now,
153 updatedAt: now,
154 archivedAt: lifecycle === "archived" ? now : null,
155 deletedAt: lifecycle === "deleted" ? now : null,
156 settledOverride: null,
157 settledAt: null,
158 lastVisitedAt: null,
159 };
160 yield* sink.write({
161 events: [
162 {
163 id: EventId.make(`${threadId}:created`),
164 type: "thread.created",
165 threadId,
166 occurredAt: now,
167 payload: thread,
168 },
169 {
170 id: EventId.make(`${threadId}:provider`),
171 type: "provider-thread.updated",
172 threadId,
173 occurredAt: now,
174 payload: providerThread,
175 },
176 ],
177 });
178 return threadId;
179});
180
181const makeImporter = Effect.fn(function* (
182 candidates: ReadonlyArray<ReturnType<typeof candidate>>,
183 beforeTransaction: Effect.Effect<void> = Effect.void,
184 getAdapter?: ProviderAdapterRegistry.ProviderAdapterRegistryV2Shape["get"],
185 threadCommands?: KeyedLock.KeyedLock<ThreadId>,
186 settings: Layer.Layer<
187 ServerSettings.ServerSettingsService,
188 ServerSettingsError
189 > = ServerSettings.layerTest(),
190) {
191 const sql = yield* SqlClient.SqlClient;
192 const events = yield* EventStore.EventStoreV2;
193 const projections = yield* ProjectionStore.ProjectionStoreV2;
194 const published: Array<ApplicationStoredEvent> = [];
195 const withTransaction: typeof sql.withTransaction = (effect) =>
196 beforeTransaction.pipe(Effect.andThen(sql.withTransaction(effect)));
197 const importSql = new Proxy(sql, {
198 get: (target, key, receiver) =>
199 key === "withTransaction" ? withTransaction : Reflect.get(target, key, receiver),
200 });
201 const sinkLayer = Layer.fresh(EventSink.layer).pipe(
202 Layer.provide(
203 Layer.mergeAll(
204 Layer.succeed(SqlClient.SqlClient, importSql),
205 Layer.succeed(ProjectionStore.ProjectionStoreV2, projections),
206 Layer.succeed(EventStore.EventStoreV2, {
207 ...events,
208 publishCommitted: (batch) =>
209 events.publishCommitted(batch).pipe(
210 Effect.tap(() =>
211 Effect.sync(() => {
212 published.push(...batch);
213 }),
214 ),
215 ),
216 }),
217 ),
218 ),
219 );
220 const context = yield* Layer.build(
221 AgentSessionImporter.layer.pipe(
222 Layer.provide(
223 Layer.mergeAll(
224 sinkLayer,
225 settings,
226 Layer.succeed(IdAllocator.IdAllocatorV2, yield* IdAllocator.IdAllocatorV2),
227 Layer.succeed(
228 ProviderSessionRuntime.ProviderSessionRuntimeRepository,
229 yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository,
230 ),
231 Layer.mock(AgentSessionScanner.AgentSessionScanner)({
232 recentThreads: () => Stream.fromIterable(candidates),
233 }),
234 Layer.mock(ProjectService.ProjectService)({
235 getById: () =>
236 Effect.succeedSome({
237 id: projectId,
238 title: "Project",
239 workspaceRoot: "/workspace/project",
240 defaultModelSelection: null,
241 scripts: [],
242 createdAt: timestamp,
243 updatedAt: timestamp,
244 deletedAt: null,
245 }),
246 }),
247 Layer.mock(Orchestrator.OrchestratorV2)({
248 getThreadRecords: (threadId, fields) =>
249 projections
250 .getThreadRecords(threadId, fields)
251 .pipe(
252 Effect.mapError(
253 (cause) => new Orchestrator.OrchestratorProjectionError({ threadId, cause }),
254 ),
255 ),
256 }),
257 Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)(
258 getAdapter === undefined ? {} : { get: getAdapter },
259 ),
260 Layer.succeed(
261 ThreadCommandExecutor.ThreadCommandExecutor,
262 threadCommands ?? (yield* KeyedLock.make<ThreadId>()),
263 ),
264 ),
265 ),
266 ),
267 );
268 const importer = Context.get(context, AgentSessionImporter.AgentSessionImporter);
269 return {
270 run: importer.importRecentAgentThreads({ projectId }),
271 importNativeHistory: importer.importNativeHistory,
272 published,
273 };
274});
275
276it.effect(
277 "ignores owned sessions, including archived and deleted threads, but imports other instances and unowned sessions",
278 () =>
279 Effect.gen(function* () {
280 const projections = yield* ProjectionStore.ProjectionStoreV2;
281 const events = yield* EventStore.EventStoreV2;
282 const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
283 const active = candidate(CODEX_MODEL_SELECTION, "active");
284 const archived = candidate(
285 CLAUDE_MODEL_SELECTION,
286 "020e7f00-dce6-406b-8ec6-ed90498945d6",
287 "claudeAgent",
288 );
289 const deleted = candidate(CODEX_MODEL_SELECTION, "deleted");
290 const ownedPi = candidate(PI_MODEL_SELECTION, "owned-pi", "pi");
291 const owners = [
292 yield* seedOwner(active),
293 yield* seedOwner(archived, "archived"),
294 yield* seedOwner(deleted, "deleted"),
295 yield* seedOwner(ownedPi),
296 ];
297 const before = yield* Effect.forEach(owners, (id) =>
298 projections.getThreadRecords(id, ["providerThreads"]),
299 );
300 const otherInstance = candidate(
301 { ...CLAUDE_MODEL_SELECTION, instanceId: ProviderInstanceId.make("claude-other") },
302 archived.source.providerSessionId,
303 "claudeAgent",
304 );
305 const unowned = candidate(CODEX_MODEL_SELECTION, "unowned");
306 unowned.thread.omittedMessageCount = 1;
307 const unownedPi = candidate(PI_MODEL_SELECTION, "unowned-pi", "pi");
308 const importer = yield* makeImporter([
309 active,
310 archived,
311 deleted,
312 ownedPi,
313 otherInstance,
314 unowned,
315 unownedPi,
316 ]);
317
318 expect(yield* importer.run).toEqual({ importedCount: 3, skippedCount: 0 });
319 for (const source of [active, archived, deleted, ownedPi]) {
320 const threadId = importId(source);
321 expect(yield* Effect.option(projections.getThread(threadId))).toEqual(Option.none());
322 expect(yield* events.read({ threadId }).pipe(Stream.runCollect)).toEqual([]);
323 expect(yield* runtimes.getByThreadId({ threadId })).toEqual(Option.none());
324 }
325 expect(
326 yield* Effect.forEach(owners, (id) =>
327 projections.getThreadRecords(id, ["providerThreads"]),
328 ),
329 ).toEqual(before);
330 expect(
331 new Set(
332 importer.published.flatMap((stored) =>
333 "event" in stored ? [stored.event.threadId] : [],
334 ),
335 ),
336 ).toEqual(new Set([importId(otherInstance), importId(unowned), importId(unownedPi)]));
337 for (const source of [otherInstance, unowned, unownedPi]) {
338 const threadId = importId(source);
339 const records = yield* projections.getThreadRecords(threadId, [
340 "messages",
341 "turnItems",
342 "providerThreads",
343 ]);
344 expect(records.messages.map((message) => message.text)).toEqual(["Fix it", "Fixed"]);
345 const notices = records.turnItems.filter((item) => item.type === "system_notice");
346 if (source === unowned) {
347 expect(records.turnItems[0]).toBe(notices[0]);
348 expect(notices).toMatchObject([
349 {
350 ordinal: 1,
351 runId: null,
352 message: expect.stringContaining("1 earlier message was not imported"),
353 },
354 ]);
355 } else {
356 expect(notices).toEqual([]);
357 }
358 expect(records.thread.historyOrigin).toBe("v1_import");
359 expect(records.thread.activeProviderThreadId).toBe(records.providerThreads[0]?.id);
360 expect(records.providerThreads[0]?.nativeThreadRef?.nativeId).toBe(nativeId(source));
361 expect(Option.getOrThrow(yield* runtimes.getByThreadId({ threadId }))).toMatchObject({
362 providerInstanceId: source.source.providerInstanceId,
363 resumeCursor:
364 source.source.provider === "claudeAgent"
365 ? { threadId, resume: source.source.providerSessionId }
366 : { threadId: nativeId(source) },
367 runtimePayload: { importedTranscripts: [source.source] },
368 });
369 }
370 }).pipe(Effect.scoped, Effect.provide(persistence)),
371);
372
373it.effect("rechecks ownership after the import reaches the commit transaction", () =>
374 Effect.gen(function* () {
375 const enteringTransaction = yield* Deferred.make<void>();
376 const continueTransaction = yield* Deferred.make<void>();
377 const source = candidate(CODEX_MODEL_SELECTION, "claimed-during-import");
378 const importer = yield* makeImporter(
379 [source],
380 Deferred.succeed(enteringTransaction, undefined).pipe(
381 Effect.andThen(Deferred.await(continueTransaction)),
382 ),
383 );
384 const pending = yield* Effect.forkScoped(importer.run);
385 yield* Deferred.await(enteringTransaction);
386 // Commit ownership after all pre-transaction reads, before the import acquires SQLite.
387 const ownerId = yield* seedOwner(source);
388 yield* Deferred.succeed(continueTransaction, undefined);
389 expect(yield* Fiber.join(pending)).toEqual({ importedCount: 0, skippedCount: 0 });
390 expect(yield* importer.run).toEqual({ importedCount: 0, skippedCount: 0 });
391 const projections = yield* ProjectionStore.ProjectionStoreV2;
392 const events = yield* EventStore.EventStoreV2;
393 const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
394 expect(yield* Effect.option(projections.getThread(importId(source)))).toEqual(Option.none());
395 expect(yield* events.read({ threadId: importId(source) }).pipe(Stream.runCollect)).toEqual([]);
396 expect(yield* runtimes.list()).toEqual([]);
397 expect(importer.published).toEqual([]);
398 expect(
399 (yield* projections.getThreadRecords(ownerId, ["providerThreads"])).providerThreads[0]
400 ?.appThreadId,
401 ).toBe(ownerId);
402 }).pipe(Effect.scoped, Effect.provide(persistence)),
403);
404
405it.effect(
406 "retries failed commits and runtime persistence without replaying committed history",
407 () =>
408 Effect.gen(function* () {
409 const sql = yield* SqlClient.SqlClient;
410 const events = yield* EventStore.EventStoreV2;
411 const projections = yield* ProjectionStore.ProjectionStoreV2;
412 const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
413 const source = candidate(CODEX_MODEL_SELECTION, "retry");
414 const threadId = importId(source);
415 const importer = yield* makeImporter([source]);
416 yield* sql`CREATE TEMP TRIGGER fail_import BEFORE INSERT ON orchestration_events WHEN NEW.event_type = 'message.updated' BEGIN SELECT RAISE(ABORT, 'test publication failure'); END`;
417 expect(yield* importer.run).toEqual({ importedCount: 0, skippedCount: 1 });
418 expect(yield* events.read({ threadId }).pipe(Stream.runCollect)).toEqual([]);
419 expect(yield* Effect.option(projections.getThread(threadId))).toEqual(Option.none());
420 expect(yield* runtimes.list()).toEqual([]);
421 expect(importer.published).toEqual([]);
422 yield* sql`DROP TRIGGER fail_import`;
423
424 yield* sql`CREATE TEMP TRIGGER fail_runtime BEFORE INSERT ON provider_session_runtime BEGIN SELECT RAISE(ABORT, 'test runtime failure'); END`;
425 expect(yield* importer.run).toEqual({ importedCount: 0, skippedCount: 1 });
426 const committed = yield* events.read({ threadId }).pipe(Stream.runCollect);
427 expect(committed).toHaveLength(6);
428 expect(importer.published).toHaveLength(6);
429 expect(yield* runtimes.list()).toEqual([]);
430 yield* sql`DROP TRIGGER fail_runtime`;
431
432 expect(yield* importer.run).toEqual({ importedCount: 1, skippedCount: 0 });
433 expect(yield* importer.run).toEqual({ importedCount: 1, skippedCount: 0 });
434 expect(yield* events.read({ threadId }).pipe(Stream.runCollect)).toEqual(committed);
435 expect(importer.published).toHaveLength(6);
436 expect(
437 (yield* projections.getThreadRecords(threadId, ["messages"])).messages.map(
438 (message) => message.text,
439 ),
440 ).toEqual(["Fix it", "Fixed"]);
441 expect(Option.getOrThrow(yield* runtimes.getByThreadId({ threadId }))).toMatchObject({
442 providerInstanceId: source.source.providerInstanceId,
443 resumeCursor: { threadId: source.source.providerSessionId },
444 runtimePayload: { importedTranscripts: [source.source] },
445 });
446 }).pipe(Effect.scoped, Effect.provide(persistence)),
447);
448
449it.effect("does not record a changed transcript as imported without its new messages", () =>
450 Effect.gen(function* () {
451 const projections = yield* ProjectionStore.ProjectionStoreV2;
452 const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
453 const first = candidate(
454 CLAUDE_MODEL_SELECTION,
455 "6f1c2a7e-3b4d-4e5f-9a8b-1c2d3e4f5a6b",
456 "claudeAgent",
457 );
458 const threadId = importId(first);
459 expect(yield* (yield* makeImporter([first])).run).toEqual({
460 importedCount: 1,
461 skippedCount: 0,
462 });
463
464 const continued = candidate(
465 CLAUDE_MODEL_SELECTION,
466 first.source.providerSessionId,
467 "claudeAgent",
468 );
469 continued.source.size = 200;
470 continued.source.mtimeMs = 3;
471 continued.thread.messages.push(
472 { role: "user", text: "Continue", createdAt: timestamp },
473 { role: "assistant", text: "Continued", createdAt: timestamp },
474 );
475 yield* (yield* makeImporter([continued])).run;
476
477 const recorded = AgentSessionScanner.decodeImportedTranscriptPayload(
478 Option.getOrThrow(yield* runtimes.getByThreadId({ threadId })).runtimePayload,
479 ).pipe(Option.getOrThrow).importedTranscripts;
480 const texts = (yield* projections.getThreadRecords(threadId, ["messages"])).messages.map(
481 (message) => message.text,
482 );
483 expect({
484 recordsChangedTranscript: recorded?.some((source) => source.size === 200) ?? false,
485 showsNewMessages: texts.includes("Continued"),
486 }).not.toEqual({ recordsChangedTranscript: true, showsNewMessages: false });
487 }).pipe(Effect.scoped, Effect.provide(persistence)),
488);
489
490it.effect(
491 "copies native session history into the imported thread once, retrying a failed read",
492 () =>
493 Effect.gen(function* () {
494 const projections = yield* ProjectionStore.ProjectionStoreV2;
495 const source = candidate(CODEX_MODEL_SELECTION, "native-history");
496 const threadId = yield* seedOwner(source);
497 const resolved: Array<ProviderInstanceId> = [];
498 const opened: Array<ProviderAdapter.ProviderAdapterV2OpenSessionInput> = [];
499 let closed = false;
500 let failRead = true;
501 const replayed = (role: "user" | "assistant", text: string) => ({
502 createdBy: role === "user" ? ("user" as const) : ("agent" as const),
503 creationSource: "provider" as const,
504 id: MessageId.make(`replayed:${text}`),
505 threadId,
506 runId: null,
507 nodeId: null,
508 role,
509 text,
510 attachments: [],
511 streaming: false,
512 createdAt: now,
513 updatedAt: now,
514 });
515 const adapter = {
516 openSession: (input: ProviderAdapter.ProviderAdapterV2OpenSessionInput) =>
517 Effect.acquireRelease(
518 Effect.sync(() => {
519 opened.push(input);
520 return {
521 readThreadSnapshot: ({ providerThread }) =>
522 failRead
523 ? Effect.fail(
524 new ProviderAdapter.ProviderAdapterReadThreadSnapshotError({
525 driver: providerThread.driver,
526 providerThreadId: providerThread.id,
527 }),
528 )
529 : Effect.succeed({
530 providerThread,
531 providerTurns: [],
532 runtimeRequests: [],
533 reasoningMessageIds: [MessageId.make("replayed:Check the build log")],
534 messages: [
535 replayed("user", "Fix the build"),
536 replayed("assistant", "Check the build log"),
537 replayed("assistant", "Fixed it"),
538 replayed("assistant", " "),
539 ],
540 }),
541 } as Pick<ProviderAdapter.ProviderAdapterV2SessionRuntime, "readThreadSnapshot">;
542 }),
543 () =>
544 Effect.sync(() => {
545 closed = true;
546 }),
547 ),
548 } as unknown as ProviderAdapter.ProviderAdapterV2["Service"];
549 const importer = yield* makeImporter([], Effect.void, (instanceId) =>
550 Effect.sync(() => {
551 resolved.push(instanceId);
552 return adapter;
553 }),
554 );
555
556 const importHistory = importer.importNativeHistory({
557 threadId,
558 workspaceRoot: "/workspace/project",
559 });
560
561 expect((yield* Effect.flip(importHistory))._tag).toBe(
562 "ProviderAdapterReadThreadSnapshotError",
563 );
564 expect(closed).toBe(true);
565 expect(
566 (yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin,
567 ).toBeUndefined();
568
569 failRead = false;
570 expect(yield* importHistory).toBe(3);
571 expect(resolved).toEqual([
572 CODEX_MODEL_SELECTION.instanceId,
573 CODEX_MODEL_SELECTION.instanceId,
574 ]);
575 expect(opened[1]).toMatchObject({
576 threadId,
577 initialNativeThreadId: source.source.providerSessionId,
578 runtimePolicy: { cwd: "/workspace/project" },
579 });
580 // Forks keep run-less history only for this origin.
581 expect((yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin).toBe(
582 "v1_import",
583 );
584 expect(yield* importHistory).toBe(0);
585 expect(opened).toHaveLength(2);
586 const page = yield* projections.getTimelinePage(threadId, { view: "activity", limit: 10 });
587 expect(
588 page.items.map(({ item }) => [
589 item.type,
590 item.runId,
591 "text" in item ? item.text : undefined,
592 ]),
593 ).toEqual([
594 ["user_message", null, "Fix the build"],
595 ["reasoning", null, "Check the build log"],
596 ["assistant_message", null, "Fixed it"],
597 ]);
598 }).pipe(Effect.scoped, Effect.provide(persistence)),
599);
600
601it.effect("releases the ACP import lock when a native session never answers", () =>
602 Effect.gen(function* () {
603 const projections = yield* ProjectionStore.ProjectionStoreV2;
604 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
605 const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, "native-history-hang"));
606 const opened = yield* Deferred.make<void>();
607 let closed = false;
608 // An agent that only resumes sessions and never answers its first session/resume.
609 const adapter = {
610 openSession: () =>
611 Effect.suspend(() => {
612 if (!closed) {
613 return Effect.addFinalizer(() =>
614 Effect.sync(() => {
615 closed = true;
616 }),
617 ).pipe(
618 Effect.andThen(Deferred.succeed(opened, undefined)),
619 Effect.andThen(Effect.never),
620 );
621 }
622 return Effect.succeed({
623 readThreadSnapshot: ({ providerThread }) =>
624 Effect.succeed({
625 providerThread,
626 providerTurns: [],
627 runtimeRequests: [],
628 messages: [
629 {
630 createdBy: "user",
631 creationSource: "provider",
632 id: MessageId.make("replayed:hang"),
633 threadId,
634 runId: null,
635 nodeId: null,
636 role: "user",
637 text: "Fix the build",
638 attachments: [],
639 streaming: false,
640 createdAt: now,
641 updatedAt: now,
642 },
643 ],
644 }),
645 } as Pick<ProviderAdapter.ProviderAdapterV2SessionRuntime, "readThreadSnapshot">);
646 }),
647 } as unknown as ProviderAdapter.ProviderAdapterV2["Service"];
648 const importer = yield* makeImporter([], Effect.void, () => Effect.succeed(adapter));
649 const importHistory = coordinator.withSessionMutation(
650 importer.importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }),
651 );
652
653 const importing = yield* importHistory.pipe(Effect.flip, Effect.forkChild);
654 yield* Deferred.await(opened);
655 // Any other ACP import or delete waits for the same lock.
656 const next = yield* coordinator
657 .withSessionMutation(Effect.succeed("next"))
658 .pipe(Effect.forkChild);
659 yield* TestClock.adjust("2 minutes");
660
661 expect((yield* Fiber.join(importing))._tag).toBe("TimeoutError");
662 expect(closed).toBe(true);
663 expect(yield* Fiber.join(next)).toBe("next");
664 expect(
665 (yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin,
666 ).toBeUndefined();
667 expect(yield* importHistory).toBe(1);
668 expect((yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin).toBe(
669 "v1_import",
670 );
671 }).pipe(
672 Effect.scoped,
673 Effect.provide(Layer.merge(persistence, AcpRegistryRuntimeCoordinator.layer)),
674 ),
675);
676
677/**
678 * Runs a thread command that read the thread before the history load ended and
679 * writes once the importer is about to commit, as an unserialized write would.
680 */
681const importDuringCommand = Effect.fn(function* (
682 name: string,
683 command: (thread: OrchestrationV2AppThread) => ReadonlyArray<OrchestrationV2DomainEvent>,
684) {
685 const projections = yield* ProjectionStore.ProjectionStoreV2;
686 const sink = yield* EventSink.EventSinkV2;
687 const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, name));
688 const lock = yield* KeyedLock.make<ThreadId>();
689 const loaded = yield* Deferred.make<void>();
690 const importerCommitting = yield* Deferred.make<void>();
691 const commandWritten = yield* Deferred.make<void>();
692 const committing = Deferred.succeed(importerCommitting, undefined);
693 const adapter = {
694 openSession: () =>
695 Effect.succeed({
696 readThreadSnapshot: ({ providerThread }) =>
697 Deferred.await(loaded).pipe(
698 Effect.as({
699 providerThread,
700 providerTurns: [],
701 runtimeRequests: [],
702 messages: [
703 {
704 createdBy: "user" as const,
705 creationSource: "provider" as const,
706 id: MessageId.make(`replayed:${name}`),
707 threadId,
708 runId: null,
709 nodeId: null,
710 role: "user" as const,
711 text: "Fix the build",
712 attachments: [],
713 streaming: false,
714 createdAt: now,
715 updatedAt: now,
716 },
717 ],
718 }),
719 ),
720 } as Pick<ProviderAdapter.ProviderAdapterV2SessionRuntime, "readThreadSnapshot">),
721 } as unknown as ProviderAdapter.ProviderAdapterV2["Service"];
722 const importer = yield* makeImporter(
723 [],
724 committing.pipe(Effect.andThen(Deferred.await(commandWritten))),
725 () => Effect.succeed(adapter),
726 {
727 ...lock,
728 withLock: (key, effect) => committing.pipe(Effect.andThen(lock.withLock(key, effect))),
729 },
730 );
731 const commanding = yield* Effect.forkScoped(
732 lock.withLock(
733 threadId,
734 Effect.gen(function* () {
735 const { thread } = yield* projections.getThreadRecords(threadId, []);
736 yield* Deferred.succeed(loaded, undefined);
737 yield* Deferred.await(importerCommitting);
738 yield* sink.write({ events: command(thread) });
739 yield* Deferred.succeed(commandWritten, undefined);
740 }),
741 ),
742 );
743 const imported = yield* importer.importNativeHistory({
744 threadId,
745 workspaceRoot: "/workspace/project",
746 });
747 yield* Fiber.join(commanding);
748 const records = yield* projections.getThreadRecords(threadId, ["runs"]);
749 const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 });
750 return { imported, thread: records.thread, runs: records.runs, items: page.items };
751});
752
753it.effect("keeps a thread edit made while the native history loads", () =>
754 Effect.gen(function* () {
755 const result = yield* importDuringCommand("renamed-during-load", (thread) => [
756 {
757 id: EventId.make(`${thread.id}:renamed`),
758 type: "thread.metadata-updated",
759 threadId: thread.id,
760 occurredAt: now,
761 payload: { ...thread, title: "Renamed" },
762 },
763 ]);
764 expect(result.imported).toBe(1);
765 expect(result.thread.title).toBe("Renamed");
766 expect(result.thread.historyOrigin).toBe("v1_import");
767 expect(result.items).toHaveLength(1);
768 }).pipe(Effect.scoped, Effect.provide(persistence)),
769);
770
771it.effect("skips the native history when a run starts while it loads", () =>
772 Effect.gen(function* () {
773 const result = yield* importDuringCommand("run-during-load", (thread) => [
774 {
775 id: EventId.make(`${thread.id}:run`),
776 type: "run.updated",
777 threadId: thread.id,
778 occurredAt: now,
779 payload: {
780 id: RunId.make(`${thread.id}:run`),
781 threadId: thread.id,
782 ordinal: 1,
783 providerInstanceId: thread.providerInstanceId,
784 modelSelection: thread.modelSelection,
785 providerThreadId: null,
786 userMessageId: MessageId.make(`${thread.id}:prompt`),
787 rootNodeId: null,
788 activeAttemptId: null,
789 status: "queued",
790 requestedAt: now,
791 startedAt: null,
792 completedAt: null,
793 checkpointId: null,
794 contextHandoffId: null,
795 },
796 },
797 ]);
798 expect(result.imported).toBe(0);
799 expect(result.runs).toHaveLength(1);
800 expect(result.thread.historyOrigin).toBeUndefined();
801 expect(result.items).toEqual([]);
802 }).pipe(Effect.scoped, Effect.provide(persistence)),
803);
804
805it.effect("imports under a policy ceiling in the project's default, kept under the ceiling", () =>
806 Effect.gen(function* () {
807 const projections = yield* ProjectionStore.ProjectionStoreV2;
808 const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
809 const source = candidate(CODEX_MODEL_SELECTION, "ceiling");
810 const threadId = importId(source);
811 const importer = yield* makeImporter(
812 [source],
813 Effect.void,
814 undefined,
815 undefined,
816 ServerSettings.layerTest(
817 { defaultRuntimeMode: "full-access" },
818 {
819 organization: "Example Corp",
820 lockedSettings: [],
821 limits: { maxRuntimeMode: "auto-accept-edits" },
822 },
823 ),
824 );
825 expect(yield* importer.run).toEqual({ importedCount: 1, skippedCount: 0 });
826 expect((yield* projections.getThread(threadId)).runtimeMode).toBe("auto-accept-edits");
827 expect(Option.getOrThrow(yield* runtimes.getByThreadId({ threadId })).runtimeMode).toBe(
828 "auto-accept-edits",
829 );
830 }).pipe(Effect.scoped, Effect.provide(persistence)),
831);