1import {
2 DEFAULT_MODEL,
3 DEFAULT_MODEL_BY_PROVIDER,
4 DEFAULT_PROVIDER_INTERACTION_MODE,
5 DEFAULT_RUNTIME_MODE,
6 AgentSessionImportProjectChangedError,
7 AgentSessionImportProjectNotFoundError,
8 AgentSessionImportSource,
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 * as Context from "effect/Context";
27import * as DateTime from "effect/DateTime";
28import * as Effect from "effect/Effect";
29import * as Layer from "effect/Layer";
30import * as Option from "effect/Option";
31import * as Schema from "effect/Schema";
32import * as Stream from "effect/Stream";
33
34import * as EventSink from "../orchestration-v2/EventSink.ts";
35import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
36import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
37import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts";
38import * as AgentSessionScanner from "./AgentSessionScanner.ts";
39import * as ProjectService from "./ProjectService.ts";
40
41const IMPORT_EVENT_PREFIX = "agent-session-import:v2";
42const CLAUDE_SESSION_ID_PATTERN =
43 /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
44const decodeImportedTranscriptPayload = Schema.decodeUnknownOption(
45 Schema.Struct({
46 cwd: Schema.optional(Schema.String),
47 importedTranscripts: Schema.optional(Schema.Array(AgentSessionImportSource)),
48 }),
49);
50
51class AgentSessionUnresumableSessionError extends Schema.TaggedError<AgentSessionUnresumableSessionError>()(
52 "AgentSessionUnresumableSessionError",
53 {
54 source: AgentSessionSource,
55 providerSessionId: Schema.String,
56 },
57) {
58 override get message(): string {
59 return `Session '${this.providerSessionId}' from '${this.source}' cannot be resumed.`;
60 }
61}
62
63class AgentSessionThreadProjectConflictError extends Schema.TaggedError<AgentSessionThreadProjectConflictError>()(
64 "AgentSessionThreadProjectConflictError",
65 {
66 threadId: ThreadId,
67 expectedProjectId: ProjectId,
68 actualProjectId: ProjectId,
69 },
70) {
71 override get message(): string {
72 return `Imported thread '${this.threadId}' belongs to project '${this.actualProjectId}', not '${this.expectedProjectId}'.`;
73 }
74}
75
76class AgentSessionThreadModifiedError extends Schema.TaggedError<AgentSessionThreadModifiedError>()(
77 "AgentSessionThreadModifiedError",
78 { threadId: ThreadId },
79) {
80 override get message(): string {
81 return `Imported thread '${this.threadId}' already contains non-imported activity.`;
82 }
83}
84
85function dateTime(value: string): DateTime.Utc {
86 return DateTime.makeUnsafe(value);
87}
88
89function messageEvents(input: {
90 readonly threadId: ThreadId;
91 readonly index: number;
92 readonly message: AgentSessionScanner.AgentSessionThreadMessage;
93}): ReadonlyArray<OrchestrationV2DomainEvent> {
94 const ordinal = input.index + 1;
95 const suffix = String(input.index).padStart(6, "0");
96 const messageId = MessageId.make(`${input.threadId}:${suffix}`);
97 const turnItemId = TurnItemId.make(
98 `${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:${suffix}`,
99 );
100 const at = dateTime(input.message.createdAt);
101 const message: OrchestrationV2ConversationMessage = {
102 createdBy: input.message.role === "user" ? "user" : "agent",
103 creationSource: "server",
104 id: messageId,
105 threadId: input.threadId,
106 runId: null,
107 nodeId: null,
108 role: input.message.role,
109 text: input.message.text,
110 attachments: [],
111 streaming: false,
112 createdAt: at,
113 updatedAt: at,
114 };
115 const common = {
116 id: turnItemId,
117 threadId: input.threadId,
118 runId: null,
119 nodeId: null,
120 providerThreadId: null,
121 providerTurnId: null,
122 nativeItemRef: null,
123 parentItemId: null,
124 ordinal,
125 status: "completed" as const,
126 title: null,
127 startedAt: at,
128 completedAt: at,
129 updatedAt: at,
130 };
131 const turnItem: OrchestrationV2TurnItem =
132 input.message.role === "user"
133 ? {
134 ...common,
135 createdBy: "user",
136 creationSource: "server",
137 type: "user_message",
138 messageId,
139 inputIntent: "turn_start",
140 text: input.message.text,
141 attachments: [],
142 }
143 : {
144 ...common,
145 type: "assistant_message",
146 messageId,
147 text: input.message.text,
148 streaming: false,
149 };
150 return [
151 {
152 id: EventId.make(`${IMPORT_EVENT_PREFIX}:message:${input.threadId}:${suffix}`),
153 type: "message.updated",
154 threadId: input.threadId,
155 occurredAt: at,
156 payload: message,
157 },
158 {
159 id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:${suffix}`),
160 type: "turn-item.updated",
161 threadId: input.threadId,
162 occurredAt: at,
163 payload: turnItem,
164 },
165 ];
166}
167
168const make = Effect.gen(function* () {
169 const scanner = yield* AgentSessionScanner.AgentSessionScanner;
170 const orchestrator = yield* Orchestrator.OrchestratorV2;
171 const projects = yield* ProjectService.ProjectService;
172 const eventSink = yield* EventSink.EventSinkV2;
173 const idAllocator = yield* IdAllocator.IdAllocatorV2;
174 const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
175 const importRecentAgentThreads = Effect.fn("importRecentAgentThreadsV2")(function* (
176 input: AgentSessionImportInput,
177 ) {
178 const project = yield* projects.getById(input.projectId).pipe(
179 Effect.mapError((cause) => new AgentSessionScanError({ operation: "read-projects", cause })),
180 Effect.flatMap(
181 Option.match({
182 onNone: () =>
183 Effect.fail(new AgentSessionImportProjectNotFoundError({ projectId: input.projectId })),
184 onSome: Effect.succeed,
185 }),
186 ),
187 );
188 if (
189 input.expectedWorkspaceRoot !== undefined &&
190 normalizeProjectPathForComparison(project.workspaceRoot) !==
191 normalizeProjectPathForComparison(input.expectedWorkspaceRoot)
192 ) {
193 return yield* new AgentSessionImportProjectChangedError({ projectId: input.projectId });
194 }
195 const runtimeRows = yield* runtimes
196 .list()
197 .pipe(
198 Effect.mapError(
199 (cause) => new AgentSessionScanError({ operation: "read-projects", cause }),
200 ),
201 );
202 const completedSources = runtimeRows.flatMap((runtime) => {
203 const payload = decodeImportedTranscriptPayload(runtime.runtimePayload);
204 if (
205 Option.isNone(payload) ||
206 payload.value.cwd === undefined ||
207 normalizeProjectPathForComparison(payload.value.cwd) !==
208 normalizeProjectPathForComparison(project.workspaceRoot)
209 ) {
210 return [];
211 }
212 return payload.value.importedTranscripts ?? [];
213 });
214 const outcomes = scanner.recentThreads(project.workspaceRoot, completedSources);
215 const importedThreadIds = new Set<ThreadId>();
216 let importedCount = 0;
217 let skippedCount = 0;
218
219 yield* Stream.runForEach(outcomes, (outcome) =>
220 Effect.gen(function* () {
221 if (outcome._tag === "Skipped") {
222 skippedCount += 1;
223 return;
224 }
225 const source = outcome.source;
226 const threadId = ThreadId.make(
227 `import:${source.providerInstanceId}:${source.providerSessionId}`,
228 );
229 if (outcome._tag === "AlreadyImported") {
230 importedThreadIds.add(threadId);
231 importedCount += 1;
232 return;
233 }
234 if (outcome._tag === "Duplicate") {
235 if (importedThreadIds.has(threadId)) {
236 yield* runtimes.recordImportedTranscript({ threadId, source }).pipe(Effect.ignore);
237 }
238 return;
239 }
240
241 const imported = yield* Effect.gen(function* () {
242 const thread = outcome.thread;
243 if (
244 thread.source === "claudeAgent" &&
245 !CLAUDE_SESSION_ID_PATTERN.test(thread.providerSessionId)
246 ) {
247 return yield* new AgentSessionUnresumableSessionError({
248 source: thread.source,
249 providerSessionId: thread.providerSessionId,
250 });
251 }
252 const existing = yield* Effect.option(orchestrator.getThreadRecords(threadId, []));
253 if (Option.isSome(existing)) {
254 if (existing.value.thread.projectId !== input.projectId) {
255 return yield* new AgentSessionThreadProjectConflictError({
256 threadId,
257 expectedProjectId: input.projectId,
258 actualProjectId: existing.value.thread.projectId,
259 });
260 }
261 if (existing.value.thread.historyOrigin !== "v1_import") {
262 return yield* new AgentSessionThreadModifiedError({ threadId });
263 }
264 yield* runtimes.recordImportedTranscript({ threadId, source });
265 return true;
266 }
267
268 const driver = ProviderDriverKind.make(thread.source);
269 const model = thread.model ?? DEFAULT_MODEL_BY_PROVIDER[driver] ?? DEFAULT_MODEL;
270 const providerThreadId = idAllocator.derive.providerThread({
271 driver,
272 nativeThreadId: thread.providerSessionId,
273 });
274 const createdAt = dateTime(thread.createdAt);
275 const updatedAt = dateTime(thread.updatedAt);
276 const appThread: OrchestrationV2AppThread = {
277 createdBy: "system",
278 creationSource: "server",
279 id: threadId,
280 projectId: input.projectId,
281 title: thread.title.trim() === "" ? "Untitled thread" : thread.title,
282 providerInstanceId: thread.providerInstanceId,
283 modelSelection: { instanceId: thread.providerInstanceId, model },
284 runtimeMode: DEFAULT_RUNTIME_MODE,
285 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
286 branch: null,
287 worktreePath: null,
288 linkedPullRequest: null,
289 branchPullRequest: null,
290 activeProviderThreadId: providerThreadId,
291 historyOrigin: "v1_import",
292 lineage: {
293 parentThreadId: null,
294 relationshipToParent: null,
295 rootThreadId: threadId,
296 },
297 forkedFrom: null,
298 createdAt,
299 updatedAt,
300 archivedAt: null,
301 settledOverride: "settled",
302 settledAt: updatedAt,
303 unsettledAt: null,
304 snoozedUntil: null,
305 snoozedAt: null,
306 pinnedAt: null,
307 pinOrderKey: null,
308 activeOrderKey: null,
309 lastVisitedAt: null,
310 deletedAt: null,
311 };
312 const providerThread: OrchestrationV2ProviderThread = {
313 id: providerThreadId,
314 driver,
315 providerInstanceId: thread.providerInstanceId,
316 providerSessionId: null,
317 appThreadId: threadId,
318 ownerNodeId: null,
319 nativeThreadRef: {
320 driver,
321 nativeId: thread.providerSessionId,
322 strength: "strong",
323 },
324 nativeConversationHeadRef: null,
325 status: "idle",
326 firstRunOrdinal: null,
327 lastRunOrdinal: null,
328 handoffIds: [],
329 forkedFrom: null,
330 pendingBackgroundTasks: [],
331 createdAt,
332 updatedAt,
333 };
334
335 yield* runtimes.upsert(
336 {
337 threadId,
338 providerName: driver,
339 providerInstanceId: thread.providerInstanceId,
340 adapterKey: driver,
341 runtimeMode: DEFAULT_RUNTIME_MODE,
342 status: "stopped",
343 lastSeenAt: thread.updatedAt,
344 resumeCursor:
345 thread.source === "codex"
346 ? { threadId: thread.providerSessionId }
347 : { threadId, resume: thread.providerSessionId },
348 runtimePayload: { cwd: project.workspaceRoot },
349 },
350 { onConflict: "ignore" },
351 );
352 yield* eventSink.write({
353 events: [
354 {
355 id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${threadId}:created`),
356 type: "thread.created",
357 threadId,
358 providerInstanceId: thread.providerInstanceId,
359 occurredAt: createdAt,
360 payload: appThread,
361 },
362 ...thread.messages.flatMap((message, index) =>
363 messageEvents({ threadId, index, message }),
364 ),
365 {
366 id: EventId.make(`${IMPORT_EVENT_PREFIX}:provider-thread:${providerThreadId}`),
367 type: "provider-thread.updated",
368 threadId,
369 driver,
370 providerInstanceId: thread.providerInstanceId,
371 occurredAt: updatedAt,
372 payload: providerThread,
373 },
374 ],
375 });
376 yield* runtimes.recordImportedTranscript({ threadId, source });
377 return true;
378 }).pipe(
379 Effect.catch((cause) =>
380 Effect.logWarning("Could not import an agent session", {
381 provider: outcome.thread.source,
382 sessionId: outcome.thread.providerSessionId,
383 cause,
384 }).pipe(Effect.as(false)),
385 ),
386 );
387 if (imported) {
388 importedThreadIds.add(threadId);
389 importedCount += 1;
390 } else {
391 skippedCount += 1;
392 }
393 }),
394 );
395
396 return { importedCount, skippedCount } satisfies AgentSessionImportResult;
397 });
398
399 return { importRecentAgentThreads };
400});
401
402type AgentSessionImporterShape = Effect.Success<typeof make>;
403
404export class AgentSessionImporter extends Context.Service<
405 AgentSessionImporter,
406 AgentSessionImporterShape
407>()("t3/project/AgentSessionImporter") {}
408
409export const layer = Layer.effect(AgentSessionImporter, make);