1import {
2 DEFAULT_MODEL,
3 DEFAULT_MODEL_BY_PROVIDER,
4 DEFAULT_PROVIDER_INTERACTION_MODE,
5 DEFAULT_RUNTIME_MODE,
6 isRuntimeModeAbove,
7 AgentSessionImportProjectChangedError,
8 AgentSessionImportProjectNotFoundError,
9 AgentSessionScanError,
10 AgentSessionSource,
11 EventId,
12 MessageId,
13 ProjectId,
14 ProviderDriverKind,
15 ThreadId,
16 TurnItemId,
17 type AgentSessionImportInput,
18 type AgentSessionImportResult,
19 type OrchestrationV2AppThread,
20 type OrchestrationV2ConversationMessage,
21 type OrchestrationV2DomainEvent,
22 type OrchestrationV2ProviderThread,
23 type OrchestrationV2TurnItem,
24} from "@t3tools/contracts";
25import { normalizeProjectPathForComparison } from "@t3tools/shared/path";
26import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
27import * as Context from "effect/Context";
28import * as DateTime from "effect/DateTime";
29import * as Effect from "effect/Effect";
30import * as Layer from "effect/Layer";
31import * as Option from "effect/Option";
32import * as Schema from "effect/Schema";
33import * as Stream from "effect/Stream";
34
35import * as EventSink from "../orchestration-v2/EventSink.ts";
36import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
37import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
38import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
39import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts";
40import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts";
41import * as ServerSettings from "../serverSettings.ts";
42import * as AgentSessionScanner from "./AgentSessionScanner.ts";
43import * as ProjectService from "./ProjectService.ts";
44
45const IMPORT_EVENT_PREFIX = "agent-session-import:v2";
46const CLAUDE_SESSION_ID_PATTERN =
47 /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
48
49class AgentSessionUnresumableSessionError extends Schema.TaggedError<AgentSessionUnresumableSessionError>()(
50 "AgentSessionUnresumableSessionError",
51 {
52 source: AgentSessionSource,
53 providerSessionId: Schema.String,
54 },
55) {
56 override get message(): string {
57 return `Session '${this.providerSessionId}' from '${this.source}' cannot be resumed.`;
58 }
59}
60
61class AgentSessionThreadProjectConflictError extends Schema.TaggedError<AgentSessionThreadProjectConflictError>()(
62 "AgentSessionThreadProjectConflictError",
63 {
64 threadId: ThreadId,
65 expectedProjectId: ProjectId,
66 actualProjectId: ProjectId,
67 },
68) {
69 override get message(): string {
70 return `Imported thread '${this.threadId}' belongs to project '${this.actualProjectId}', not '${this.expectedProjectId}'.`;
71 }
72}
73
74class AgentSessionThreadModifiedError extends Schema.TaggedError<AgentSessionThreadModifiedError>()(
75 "AgentSessionThreadModifiedError",
76 { threadId: ThreadId },
77) {
78 override get message(): string {
79 return `Imported thread '${this.threadId}' already contains non-imported activity.`;
80 }
81}
82
83// ACP imports read the history while holding the session-mutation lock, and an agent can
84// leave session/resume or initialize unanswered. Longer than session/load's own deadline.
85const NATIVE_HISTORY_READ_TIMEOUT = "2 minutes";
86
87/**
88 * Whether a thread bound to a native session still lacks its native history.
89 * Once a turn has run, the native session also holds that turn's messages, so
90 * its history can no longer be imported as run-less items.
91 */
92export function isNativeHistoryPending(input: {
93 readonly historyOrigin: OrchestrationV2AppThread["historyOrigin"];
94 readonly hasRun: boolean;
95}): boolean {
96 return input.historyOrigin !== "v1_import" && !input.hasRun;
97}
98
99function dateTime(value: string): DateTime.Utc {
100 return DateTime.makeUnsafe(value);
101}
102
103type HistoryMessage = Omit<AgentSessionScanner.AgentSessionThreadMessage, "role"> & {
104 readonly role: AgentSessionScanner.AgentSessionThreadMessage["role"] | "reasoning";
105};
106
107function messageEvents(input: {
108 readonly threadId: ThreadId;
109 readonly index: number;
110 readonly message: HistoryMessage;
111}): ReadonlyArray<OrchestrationV2DomainEvent> {
112 const ordinal = input.index + 1;
113 const suffix = String(input.index).padStart(6, "0");
114 const messageId = MessageId.make(`${input.threadId}:${suffix}`);
115 const turnItemId = TurnItemId.make(
116 `${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:${suffix}`,
117 );
118 const at = dateTime(input.message.createdAt);
119 const common = {
120 id: turnItemId,
121 threadId: input.threadId,
122 runId: null,
123 nodeId: null,
124 providerThreadId: null,
125 providerTurnId: null,
126 nativeItemRef: null,
127 parentItemId: null,
128 ordinal,
129 status: "completed" as const,
130 title: null,
131 startedAt: at,
132 completedAt: at,
133 updatedAt: at,
134 };
135 const turnItemEvent = (payload: OrchestrationV2TurnItem): OrchestrationV2DomainEvent => ({
136 id: EventId.make(turnItemId),
137 type: "turn-item.updated",
138 threadId: input.threadId,
139 occurredAt: at,
140 payload,
141 });
142 if (input.message.role === "reasoning") {
143 return [
144 turnItemEvent({ ...common, type: "reasoning", text: input.message.text, streaming: false }),
145 ];
146 }
147 const message: OrchestrationV2ConversationMessage = {
148 createdBy: input.message.role === "user" ? "user" : "agent",
149 creationSource: "server",
150 id: messageId,
151 threadId: input.threadId,
152 runId: null,
153 nodeId: null,
154 role: input.message.role,
155 text: input.message.text,
156 attachments: [],
157 streaming: false,
158 createdAt: at,
159 updatedAt: at,
160 };
161 const turnItem: OrchestrationV2TurnItem =
162 input.message.role === "user"
163 ? {
164 ...common,
165 createdBy: "user",
166 creationSource: "server",
167 type: "user_message",
168 messageId,
169 inputIntent: "turn_start",
170 text: input.message.text,
171 attachments: [],
172 }
173 : {
174 ...common,
175 type: "assistant_message",
176 messageId,
177 text: input.message.text,
178 streaming: false,
179 };
180 return [
181 {
182 id: EventId.make(`${IMPORT_EVENT_PREFIX}:message:${input.threadId}:${suffix}`),
183 type: "message.updated",
184 threadId: input.threadId,
185 occurredAt: at,
186 payload: message,
187 },
188 turnItemEvent(turnItem),
189 ];
190}
191
192const make = Effect.gen(function* () {
193 const scanner = yield* AgentSessionScanner.AgentSessionScanner;
194 const orchestrator = yield* Orchestrator.OrchestratorV2;
195 const projects = yield* ProjectService.ProjectService;
196 const eventSink = yield* EventSink.EventSinkV2;
197 const idAllocator = yield* IdAllocator.IdAllocatorV2;
198 const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
199 const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
200 const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
201 const settingsService = yield* ServerSettings.ServerSettingsService;
202 const ceiling = settingsService.managedPolicy?.limits?.maxRuntimeMode;
203 const importRecentAgentThreads = Effect.fn("importRecentAgentThreadsV2")(function* (
204 input: AgentSessionImportInput,
205 ) {
206 const project = yield* projects.getById(input.projectId).pipe(
207 Effect.mapError((cause) => new AgentSessionScanError({ operation: "read-projects", cause })),
208 Effect.flatMap(
209 Option.match({
210 onNone: () =>
211 Effect.fail(new AgentSessionImportProjectNotFoundError({ projectId: input.projectId })),
212 onSome: Effect.succeed,
213 }),
214 ),
215 );
216 if (
217 input.expectedWorkspaceRoot !== undefined &&
218 normalizeProjectPathForComparison(project.workspaceRoot) !==
219 normalizeProjectPathForComparison(input.expectedWorkspaceRoot)
220 ) {
221 return yield* new AgentSessionImportProjectChangedError({ projectId: input.projectId });
222 }
223 // Imports run in full access, except under a policy ceiling: then in the project's default,
224 // which the policy keeps under it, so imported threads never sit above the ceiling.
225 const runtimeMode =
226 ceiling === undefined
227 ? DEFAULT_RUNTIME_MODE
228 : yield* settingsService.getSettings.pipe(
229 Effect.mapError(
230 (cause) => new AgentSessionScanError({ operation: "read-settings", cause }),
231 ),
232 Effect.map((settings) => {
233 const mode = resolveProjectSettings(settings, input.projectId, project).settings
234 .defaultRuntimeMode;
235 return isRuntimeModeAbove(mode, ceiling) ? ceiling : mode;
236 }),
237 );
238 const runtimeRows = yield* runtimes
239 .list()
240 .pipe(
241 Effect.mapError(
242 (cause) => new AgentSessionScanError({ operation: "read-projects", cause }),
243 ),
244 );
245 const completedSources = runtimeRows.flatMap((runtime) => {
246 const payload = AgentSessionScanner.decodeImportedTranscriptPayload(runtime.runtimePayload);
247 if (
248 Option.isNone(payload) ||
249 payload.value.cwd === undefined ||
250 normalizeProjectPathForComparison(payload.value.cwd) !==
251 normalizeProjectPathForComparison(project.workspaceRoot)
252 ) {
253 return [];
254 }
255 return payload.value.importedTranscripts ?? [];
256 });
257 const outcomes = scanner.recentThreads(project.workspaceRoot, completedSources);
258 const importedThreadIds = new Set<ThreadId>();
259 let importedCount = 0;
260 let skippedCount = 0;
261
262 yield* Stream.runForEach(outcomes, (outcome) =>
263 Effect.gen(function* () {
264 if (outcome._tag === "Skipped") {
265 skippedCount += 1;
266 return;
267 }
268 const source = outcome.source;
269 const threadId = ThreadId.make(
270 `import:${source.providerInstanceId}:${source.providerSessionId}`,
271 );
272 if (outcome._tag === "AlreadyImported") {
273 importedThreadIds.add(threadId);
274 importedCount += 1;
275 return;
276 }
277 if (outcome._tag === "Duplicate") {
278 if (importedThreadIds.has(threadId)) {
279 yield* runtimes.recordImportedTranscript({ threadId, source }).pipe(Effect.ignore);
280 }
281 return;
282 }
283
284 const imported = yield* Effect.gen(function* () {
285 const thread = outcome.thread;
286 if (
287 thread.source === "claudeAgent" &&
288 !CLAUDE_SESSION_ID_PATTERN.test(thread.providerSessionId)
289 ) {
290 return yield* new AgentSessionUnresumableSessionError({
291 source: thread.source,
292 providerSessionId: thread.providerSessionId,
293 });
294 }
295 const driver = ProviderDriverKind.make(thread.source);
296 // Pi resumes by session file path; its session UUID only names the thread.
297 const nativeThreadId =
298 thread.source === "pi" ? source.filePath : thread.providerSessionId;
299 const persistRuntime = runtimes.upsert(
300 {
301 threadId,
302 providerName: driver,
303 providerInstanceId: thread.providerInstanceId,
304 adapterKey: driver,
305 runtimeMode,
306 status: "stopped",
307 lastSeenAt: thread.updatedAt,
308 resumeCursor:
309 thread.source === "claudeAgent"
310 ? { threadId, resume: thread.providerSessionId }
311 : { threadId: nativeThreadId },
312 runtimePayload: { cwd: project.workspaceRoot },
313 },
314 { onConflict: "ignore" },
315 );
316 const existing = yield* Effect.option(orchestrator.getThreadRecords(threadId, []));
317 if (Option.isSome(existing)) {
318 if (existing.value.thread.projectId !== input.projectId) {
319 return yield* new AgentSessionThreadProjectConflictError({
320 threadId,
321 expectedProjectId: input.projectId,
322 actualProjectId: existing.value.thread.projectId,
323 });
324 }
325 if (existing.value.thread.historyOrigin !== "v1_import") {
326 return yield* new AgentSessionThreadModifiedError({ threadId });
327 }
328 // Other recorded transcripts mean this one changed after the import, and the
329 // thread does not hold its new messages, so it must stay unrecorded.
330 const recorded = Option.flatMap(yield* runtimes.getByThreadId({ threadId }), (row) =>
331 AgentSessionScanner.decodeImportedTranscriptPayload(row.runtimePayload),
332 ).pipe(Option.flatMapNullishOr((payload) => payload.importedTranscripts));
333 if (
334 Option.isSome(recorded) &&
335 recorded.value.length > 0 &&
336 !recorded.value.some((imported) =>
337 AgentSessionScanner.sameTranscriptIdentity(imported, source),
338 )
339 ) {
340 return false;
341 }
342 yield* persistRuntime;
343 yield* runtimes.recordImportedTranscript({ threadId, source });
344 return true;
345 }
346
347 const model = thread.model ?? DEFAULT_MODEL_BY_PROVIDER[driver] ?? DEFAULT_MODEL;
348 const providerThreadId = idAllocator.derive.providerThread({
349 driver,
350 providerInstanceId: thread.providerInstanceId,
351 nativeThreadId,
352 });
353 const createdAt = dateTime(thread.createdAt);
354 const updatedAt = dateTime(thread.updatedAt);
355 const appThread: OrchestrationV2AppThread = {
356 createdBy: "system",
357 creationSource: "server",
358 id: threadId,
359 projectId: input.projectId,
360 title: thread.title.trim() === "" ? "Untitled thread" : thread.title,
361 providerInstanceId: thread.providerInstanceId,
362 modelSelection: { instanceId: thread.providerInstanceId, model },
363 runtimeMode,
364 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
365 branch: null,
366 worktreePath: null,
367 linkedPullRequest: null,
368 branchPullRequest: null,
369 activeProviderThreadId: providerThreadId,
370 historyOrigin: "v1_import",
371 lineage: {
372 parentThreadId: null,
373 relationshipToParent: null,
374 rootThreadId: threadId,
375 },
376 forkedFrom: null,
377 createdAt,
378 updatedAt,
379 archivedAt: null,
380 settledOverride: "settled",
381 settledAt: updatedAt,
382 unsettledAt: null,
383 snoozedUntil: null,
384 snoozedAt: null,
385 pinnedAt: null,
386 pinOrderKey: null,
387 activeOrderKey: null,
388 lastVisitedAt: null,
389 deletedAt: null,
390 };
391 const providerThread: OrchestrationV2ProviderThread = {
392 id: providerThreadId,
393 driver,
394 providerInstanceId: thread.providerInstanceId,
395 providerSessionId: null,
396 appThreadId: threadId,
397 ownerNodeId: null,
398 nativeThreadRef: {
399 driver,
400 nativeId: nativeThreadId,
401 strength: "strong",
402 },
403 nativeConversationHeadRef: null,
404 status: "idle",
405 firstRunOrdinal: null,
406 lastRunOrdinal: null,
407 handoffIds: [],
408 forkedFrom: null,
409 pendingBackgroundTasks: [],
410 createdAt,
411 updatedAt,
412 };
413
414 const committed = yield* eventSink.write({
415 unlessNativeThreadOwned: providerThread,
416 events: [
417 {
418 id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${threadId}:created`),
419 type: "thread.created",
420 threadId,
421 providerInstanceId: thread.providerInstanceId,
422 occurredAt: createdAt,
423 payload: appThread,
424 },
425 ...(thread.omittedMessageCount > 0
426 ? [
427 {
428 id: EventId.make(`${IMPORT_EVENT_PREFIX}:history-truncated:${threadId}`),
429 type: "turn-item.updated" as const,
430 threadId,
431 occurredAt: createdAt,
432 payload: {
433 id: TurnItemId.make(`${IMPORT_EVENT_PREFIX}:history-truncated:${threadId}`),
434 threadId,
435 runId: null,
436 nodeId: null,
437 providerThreadId: null,
438 providerTurnId: null,
439 nativeItemRef: null,
440 parentItemId: null,
441 ordinal: 0,
442 status: "completed" as const,
443 title: null,
444 startedAt: createdAt,
445 completedAt: createdAt,
446 updatedAt: createdAt,
447 type: "system_notice" as const,
448 message: `${thread.omittedMessageCount} earlier ${thread.omittedMessageCount === 1 ? "message was" : "messages were"} not imported. This thread keeps the first user message and the most recent history. The original session transcript is unchanged.`,
449 },
450 },
451 ]
452 : []),
453 ...thread.messages.flatMap((message, index) =>
454 messageEvents({ threadId, index, message }),
455 ),
456 {
457 id: EventId.make(`${IMPORT_EVENT_PREFIX}:provider-thread:${providerThreadId}`),
458 type: "provider-thread.updated",
459 threadId,
460 driver,
461 providerInstanceId: thread.providerInstanceId,
462 occurredAt: updatedAt,
463 payload: providerThread,
464 },
465 ],
466 });
467 if (committed.length === 0) return null;
468 yield* persistRuntime;
469 yield* runtimes.recordImportedTranscript({ threadId, source });
470 return true;
471 }).pipe(
472 Effect.catch((cause) =>
473 Effect.logWarning("Could not import an agent session", {
474 provider: outcome.thread.source,
475 sessionId: outcome.thread.providerSessionId,
476 cause,
477 }).pipe(Effect.as(false)),
478 ),
479 );
480 if (imported === true) {
481 importedThreadIds.add(threadId);
482 importedCount += 1;
483 } else if (imported === false) {
484 skippedCount += 1;
485 }
486 }),
487 );
488
489 return { importedCount, skippedCount } satisfies AgentSessionImportResult;
490 });
491
492 /**
493 * Loading a native session replays its history only into the adapter's own
494 * snapshot, so copy that snapshot into the thread as run-less history. The
495 * `v1_import` origin keeps it in forks and marks the import done, so a call
496 * after a failed read repairs the thread and any later call does nothing.
497 */
498 const importNativeHistory = Effect.fn("importNativeHistoryV2")(function* (input: {
499 readonly threadId: ThreadId;
500 readonly workspaceRoot: string;
501 }) {
502 const pendingThread = Effect.map(
503 orchestrator.getThreadRecords(input.threadId, ["providerThreads", "runs"]),
504 (records) =>
505 isNativeHistoryPending({
506 historyOrigin: records.thread.historyOrigin,
507 hasRun: records.runs.length > 0,
508 })
509 ? records
510 : null,
511 );
512 // Thread commands write whole thread rows, so hold their lock from the
513 // re-read through the write to keep both this marker and their changes.
514 const commit = (messages: ReadonlyArray<HistoryMessage>) =>
515 threadCommands.withLock(
516 input.threadId,
517 Effect.gen(function* () {
518 const current = yield* pendingThread;
519 if (current === null) return 0;
520 yield* eventSink.write({
521 events: [
522 ...messages.flatMap((message, index) =>
523 messageEvents({ threadId: input.threadId, index, message }),
524 ),
525 {
526 id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${input.threadId}:native-history`),
527 type: "thread.metadata-updated",
528 threadId: input.threadId,
529 providerInstanceId: current.thread.providerInstanceId,
530 occurredAt: yield* DateTime.now,
531 payload: { ...current.thread, historyOrigin: "v1_import" },
532 },
533 ],
534 });
535 return messages.length;
536 }),
537 );
538 const records = yield* pendingThread;
539 if (records === null) return 0;
540 const { thread, providerThreads } = records;
541 const providerThread = providerThreads.find(
542 (candidate) => candidate.appThreadId === input.threadId && candidate.nativeThreadRef !== null,
543 );
544 const nativeThreadId = providerThread?.nativeThreadRef?.nativeId;
545 if (providerThread === undefined || nativeThreadId == null) return 0;
546 const providerSessionId = yield* idAllocator.allocate.providerSession({
547 providerInstanceId: providerThread.providerInstanceId,
548 threadId: input.threadId,
549 });
550 // The registry admits this open against sign-in changes, like any other open.
551 const adapter = yield* registry.get(thread.modelSelection.instanceId);
552 const snapshot = yield* adapter
553 .openSession({
554 threadId: input.threadId,
555 providerSessionId,
556 modelSelection: thread.modelSelection,
557 runtimePolicy: {
558 runtimeMode: thread.runtimeMode,
559 interactionMode: thread.interactionMode,
560 cwd: thread.worktreePath ?? input.workspaceRoot,
561 },
562 initialNativeThreadId: nativeThreadId,
563 ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined
564 ? {}
565 : {
566 initialProviderItemIdentityVersion: providerThread.nativeMetadata.itemIdentityVersion,
567 }),
568 })
569 .pipe(
570 Effect.flatMap((session) => session.readThreadSnapshot({ providerThread })),
571 Effect.timeout(NATIVE_HISTORY_READ_TIMEOUT),
572 Effect.scoped,
573 );
574 const reasoningIds = new Set(snapshot.reasoningMessageIds);
575 const messages = snapshot.messages.flatMap((message) =>
576 (message.role === "user" || message.role === "assistant") && message.text.trim() !== ""
577 ? [
578 {
579 role: reasoningIds.has(message.id) ? ("reasoning" as const) : message.role,
580 text: message.text,
581 createdAt: DateTime.formatIso(message.createdAt),
582 },
583 ]
584 : [],
585 );
586 // Loading the session can take a while, and the thread is usable meanwhile.
587 return yield* commit(messages);
588 });
589
590 return { importRecentAgentThreads, importNativeHistory };
591});
592
593type AgentSessionImporterShape = Effect.Success<typeof make>;
594
595export class AgentSessionImporter extends Context.Service<
596 AgentSessionImporter,
597 AgentSessionImporterShape
598>()("t3/project/AgentSessionImporter") {}
599
600export const layer = Layer.effect(AgentSessionImporter, make);