1import { ArchiveError, ProjectId } from "@t3tools/contracts";
2import * as ArchiveClient from "./ArchiveClient.ts";
3import * as HostProcess from "@t3tools/shared/HostProcess";
4import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
5import { type ArchiveManagementRequest, type ArchiveStatus, ThreadId } from "@t3tools/contracts";
6import {
7 extractMarkdownLinkHrefs,
8 resolveMarkdownFileLinkTarget,
9} from "@t3tools/shared/markdownLinks";
10import {
11 isWorkspaceImagePreviewPath,
12 audioMimeTypeFromExtension,
13 isWorkspaceVideoPreviewPath,
14} from "@t3tools/shared/filePreview";
15import * as Clock from "effect/Clock";
16import * as Context from "effect/Context";
17import * as Effect from "effect/Effect";
18import * as Exit from "effect/Exit";
19import * as FileSystem from "effect/FileSystem";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Path from "effect/Path";
23import * as Redacted from "effect/Redacted";
24import * as Schema from "effect/Schema";
25import * as Stream from "effect/Stream";
26import * as SqlClient from "effect/sql/SqlClient";
27import {
28 resolveAttachmentPathById,
29 attachmentPathsById,
30 threadHtmlRenderAttachmentIds,
31 parseAttachmentIdFromRelativePath,
32 parseThreadSegmentFromAttachmentId,
33 toSafeThreadAttachmentSegment,
34} from "../attachmentStore.ts";
35import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
36import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
37import * as ArchiveStorage from "./ArchiveStorage.ts";
38import * as ServerConfig from "../config.ts";
39
40import {
41 payloadTables,
42 Table,
43 Segment,
44 Row,
45 Metadata,
46 canonicalTables,
47 CanonicalTable,
48 Bundle,
49} from "./ArchiveFormat.ts";
50
51type LoadedBundle = Omit<typeof Bundle.Type, "segments" | "events" | "context"> & {
52 segments: typeof Metadata.Type;
53 events: ReadonlyArray<typeof Row.Type>;
54 context: ReadonlyArray<{
55 table: typeof CanonicalTable.Type;
56 rows: ReadonlyArray<typeof Row.Type>;
57 }>;
58};
59const decodeMetadata = Schema.decodeEffect(Schema.fromJsonString(Metadata));
60const decodeBundle = Schema.decodeEffect(Schema.fromJsonString(Bundle));
61const encodeBundle = Schema.encodeEffect(Schema.fromJsonString(Bundle));
62const encodeRows = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(Row)));
63const decodeRows = Schema.decodeEffect(Schema.fromJsonString(Schema.Array(Row)));
64const RecordJson = Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown));
65const decodeRecord = Schema.decodeSync(RecordJson);
66const encodeRecord = Schema.encodeSync(RecordJson);
67const encodeMetadata = Schema.encodeSync(Schema.fromJsonString(Metadata));
68const ids = {
69 messages: "message_id",
70 turn_items: "turn_item_id",
71 subagents: "subagent_id",
72} as const;
73const isArchiveError = Schema.is(ArchiveError);
74const archiveError = (cause: unknown) =>
75 isArchiveError(cause)
76 ? cause
77 : new ArchiveError({ reason: "storage", cause: Redacted.make(cause) });
78
79/** Local columns keep timeline ordering and shell state; missing bodies never decode as transcripts. */
80function metadataPayload(table: typeof Table.Type, json: string) {
81 const payload = decodeRecord(json);
82 if (table === "turn_items" && (payload.type === "error" || payload.type === "secret_request"))
83 return json;
84 const keys =
85 table === "turn_items"
86 ? [
87 "id",
88 "threadId",
89 "runId",
90 "nodeId",
91 "type",
92 "status",
93 "ordinal",
94 "messageId",
95 "inputIntent",
96 "createdBy",
97 "subagentId",
98 "requestId",
99 "createdAt",
100 "updatedAt",
101 ]
102 : table === "messages"
103 ? ["id", "threadId", "role", "createdBy", "createdAt", "updatedAt", "attachments"]
104 : [
105 "id",
106 "threadId",
107 "status",
108 "childThreadId",
109 "origin",
110 "providerThreadId",
111 "createdAt",
112 "updatedAt",
113 ];
114 return encodeRecord({
115 archivedPayload: true,
116 ...Object.fromEntries(keys.filter((key) => key in payload).map((key) => [key, payload[key]])),
117 });
118}
119
120export class TranscriptArchive extends Context.Service<
121 TranscriptArchive,
122 {
123 readonly enabled: Effect.Effect<boolean, ArchiveError>;
124 readonly manage: (
125 request: ArchiveManagementRequest,
126 ) => Effect.Effect<ArchiveStatus, ArchiveError>;
127 readonly offload: (threadId: ThreadId) => Effect.Effect<void, ArchiveError>;
128 readonly restore: (threadId: ThreadId) => Effect.Effect<void, ArchiveError>;
129 readonly read: <A, E, R>(
130 threadId: ThreadId,
131 read: Effect.Effect<A, E, R>,
132 includeHistory?: boolean,
133 ) => Effect.Effect<A, E | ArchiveError, R>;
134 readonly readStream: <A, E, R>(
135 threadId: ThreadId,
136 stream: Stream.Stream<A, E, R>,
137 ) => Stream.Stream<A, E | ArchiveError, R>;
138 readonly sweep: Effect.Effect<void, ArchiveError>;
139 readonly asset: (
140 paths: ReadonlyArray<string>,
141 ) => Effect.Effect<
142 Option.Option<{ path: string; location: string; hash: string }>,
143 ArchiveError
144 >;
145 }
146>()("t3/archive/TranscriptArchive") {}
147
148const make = Effect.gen(function* () {
149 const sql = yield* SqlClient.SqlClient;
150 const fs = yield* FileSystem.FileSystem;
151 const path = yield* Path.Path;
152 const config = yield* ServerConfig.ServerConfig;
153 const client = yield* ArchiveClient.ArchiveClient;
154 const identity = yield* ServerEnvironment.ServerEnvironmentIdentity;
155 const projects = yield* ProjectStore.ProjectStoreV2;
156 const storage = yield* ArchiveStorage.ArchiveStorage;
157 const activeThreads = sql`SELECT thread_id FROM orchestration_v2_projection_runs WHERE status IN ('preparing', 'queued', 'starting', 'running', 'waiting') UNION SELECT thread_id FROM orchestration_v2_effect_outbox WHERE status IN ('pending', 'running') UNION SELECT thread_id FROM orchestration_v2_projection_turn_items WHERE type IN ('command_execution', 'dynamic_tool', 'subagent') AND status IN ('pending', 'running', 'waiting') UNION SELECT thread_id FROM orchestration_v2_projection_runtime_requests WHERE status = 'pending'`;
158 const catalog = (threadId: ThreadId) =>
159 sql<{
160 thread_id: ThreadId;
161 location: string;
162 object_hash: string;
163 through_sequence: number;
164 }>`SELECT thread_id, location, object_hash, through_sequence FROM transcript_archives WHERE thread_id = ${threadId}`;
165 const ancestry = (threadId: ThreadId) => sql`WITH RECURSIVE ancestry(thread_id) AS (
166 SELECT ${threadId} UNION SELECT json_extract(t.payload_json, '$.forkedFrom.threadId')
167 FROM orchestration_v2_projection_threads t JOIN ancestry a ON t.thread_id = a.thread_id
168 WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run'
169 )`;
170 const dependencies = (threadId: ThreadId) =>
171 sql<{ thread_id: string }>`${ancestry(threadId)} SELECT thread_id FROM ancestry`;
172 const entries = (threadId: ThreadId, includeHistory = true) =>
173 includeHistory
174 ? sql<{
175 thread_id: ThreadId;
176 location: string;
177 object_hash: string;
178 through_sequence: number;
179 }>`${ancestry(threadId)} SELECT a.thread_id, a.location, a.object_hash, a.through_sequence FROM transcript_archives a JOIN ancestry s ON s.thread_id = a.thread_id ORDER BY a.thread_id = ${threadId} DESC`
180 : catalog(threadId);
181
182 const load = (
183 threadId: ThreadId,
184 entry: { location: string; object_hash: string; through_sequence: number },
185 ) =>
186 Effect.scoped(
187 Effect.gen(function* () {
188 const temporary = yield* fs.makeTempDirectoryScoped({ prefix: "knife-archive-read-" });
189 const file = path.join(temporary, "transcript");
190 yield* client.download(entry.location, entry.object_hash, file);
191 const manifest = yield* fs.readFileString(file).pipe(Effect.flatMap(decodeBundle));
192 const hashes = new Set([
193 manifest.events,
194 ...manifest.segments.map((row) => row.hash),
195 ...manifest.context.map((row) => row.hash),
196 ]);
197 const loaded = yield* Effect.forEach(
198 [...hashes],
199 (hash) =>
200 Effect.gen(function* () {
201 const file = path.join(temporary, hash);
202 yield* client.download(entry.location, hash, file);
203 return [
204 hash,
205 yield* fs.readFileString(file).pipe(Effect.flatMap(decodeRows)),
206 ] as const;
207 }),
208 { concurrency: 4 },
209 );
210 const chunks = new Map(loaded);
211 const segments: Array<typeof Segment.Type> = [];
212 for (const segment of manifest.segments) {
213 const rows = chunks.get(segment.hash);
214 if (!rows) return yield* new ArchiveError({ reason: "integrity" });
215 segments.push({ table: segment.table, rows });
216 }
217 const context: Array<LoadedBundle["context"][number]> = [];
218 for (const segment of manifest.context) {
219 const rows = chunks.get(segment.hash);
220 if (!rows) return yield* new ArchiveError({ reason: "integrity" });
221 context.push({ table: segment.table, rows });
222 }
223 const events = chunks.get(manifest.events);
224 if (!events) return yield* new ArchiveError({ reason: "integrity" });
225 const bundle = { ...manifest, segments, context, events };
226 if (bundle.threadId !== threadId || bundle.throughSequence !== entry.through_sequence)
227 return yield* new ArchiveError({ reason: "integrity" });
228 for (const segment of bundle.segments)
229 for (const row of segment.rows) {
230 if (
231 row.thread_id !== threadId ||
232 typeof row[ids[segment.table]] !== "string" ||
233 typeof row.payload_json !== "string"
234 )
235 return yield* new ArchiveError({ reason: "integrity" });
236 }
237 for (const event of bundle.events) {
238 if (
239 typeof event.sequence !== "number" ||
240 typeof event.payload_json !== "string" ||
241 event.stream_id !== threadId
242 )
243 return yield* new ArchiveError({ reason: "integrity" });
244 }
245 return bundle;
246 }),
247 ).pipe(Effect.mapError(archiveError));
248 const install = (bundle: LoadedBundle) =>
249 Effect.gen(function* () {
250 for (const segment of bundle.segments)
251 for (const row of segment.rows) {
252 yield* sql`UPDATE ${sql(`orchestration_v2_projection_${segment.table}`)} SET payload_json = ${row.payload_json} WHERE thread_id = ${bundle.threadId} AND ${sql(ids[segment.table])} = ${row[ids[segment.table]]}`;
253 }
254 for (const event of bundle.events)
255 yield* sql`UPDATE orchestration_events SET payload_json = ${event.payload_json} WHERE sequence = ${event.sequence} AND stream_id = ${bundle.threadId} AND payload_json = 'null'`;
256 yield* sql`DELETE FROM transcript_archives WHERE thread_id = ${bundle.threadId}`;
257 });
258 const restore = Effect.fn("TranscriptArchive.restore")(function* (threadId: ThreadId) {
259 const entry = (yield* catalog(threadId))[0];
260 if (!entry) return;
261 const bundle = yield* load(threadId, entry);
262 yield* sql.withTransaction(
263 Effect.gen(function* () {
264 const current = (yield* catalog(threadId))[0];
265 if (!current) return;
266 if (current.object_hash !== entry.object_hash)
267 return yield* new ArchiveError({ reason: "changed" });
268 for (const asset of bundle.assets) {
269 const relative = path.relative(config.attachmentsDir, asset.path);
270 if (
271 relative.startsWith("..") ||
272 path.isAbsolute(relative) ||
273 (yield* fs.exists(asset.path))
274 )
275 continue;
276 yield* Effect.scoped(
277 Effect.gen(function* () {
278 yield* fs.makeDirectory(config.attachmentsDir, { recursive: true, mode: 0o700 });
279 const temp = yield* fs.makeTempDirectoryScoped({
280 directory: config.attachmentsDir,
281 prefix: ".restore-",
282 });
283 const file = path.join(temp, "attachment");
284 yield* client.download(entry.location, asset.hash, file);
285 yield* fs.makeDirectory(path.dirname(asset.path), { recursive: true, mode: 0o700 });
286 yield* fs.open(file, { flag: "r+" }).pipe(Effect.flatMap((handle) => handle.sync));
287 yield* fs.rename(file, asset.path);
288 if (!(yield* HostProcess.isWindows))
289 yield* fs.open(config.attachmentsDir).pipe(Effect.flatMap((handle) => handle.sync));
290 }),
291 );
292 }
293 yield* install(bundle);
294 }),
295 );
296 }, Effect.mapError(archiveError));
297 const read = <A, E, R>(
298 threadId: ThreadId,
299 operation: Effect.Effect<A, E, R>,
300 includeHistory = true,
301 ) =>
302 Effect.gen(function* () {
303 const ancestors = yield* entries(threadId, includeHistory).pipe(
304 Effect.mapError(archiveError),
305 );
306 const bundles: Array<{ payload: LoadedBundle; hash: string }> = [];
307 const covered = new Set<string>();
308 for (const ancestor of ancestors) {
309 if (covered.has(ancestor.thread_id)) continue;
310 const bundle = yield* load(ancestor.thread_id, ancestor);
311 bundles.push({ payload: bundle, hash: ancestor.object_hash });
312 covered.add(ancestor.thread_id);
313 for (const segment of bundle.context)
314 for (const row of segment.rows) {
315 if (
316 segment.table === "orchestration_v2_projection_threads" &&
317 typeof row.thread_id === "string"
318 )
319 covered.add(row.thread_id);
320 }
321 }
322 if (bundles.length === 0) return yield* operation;
323 const rollback = Symbol();
324 let result: Exit.Exit<A, E> | undefined;
325 yield* sql
326 .withTransaction(
327 Effect.gen(function* () {
328 for (const { payload: bundle, hash } of bundles) {
329 const current = (yield* catalog(bundle.threadId))[0];
330 if (!current) continue;
331 if (
332 current.object_hash !== hash ||
333 current.through_sequence !== bundle.throughSequence
334 )
335 return yield* new ArchiveError({ reason: "changed" });
336 yield* install(bundle);
337 for (const segment of bundle.context)
338 for (const row of segment.rows) {
339 if (row.thread_id === bundle.threadId) continue;
340 yield* sql`INSERT OR REPLACE INTO ${sql(segment.table)} ${sql.insert(row)}`;
341 }
342 }
343 result = yield* Effect.exit(operation);
344 return yield* Effect.fail(rollback);
345 }),
346 )
347 .pipe(
348 Effect.catchIf(
349 (error) => error === rollback,
350 () => Effect.void,
351 ),
352 Effect.mapError(archiveError),
353 );
354 if (!result) return yield* Effect.die("An archived read must produce a result");
355 return yield* result;
356 });
357 const publish = Effect.fn("TranscriptArchive.publish")(function* (
358 threadId: ThreadId,
359 manifestHash: string,
360 ) {
361 const row = (yield* sql<{
362 project_id: string;
363 }>`SELECT project_id FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}`)[0];
364 if (!row) return yield* new ArchiveError({ reason: "changed" });
365 const project = yield* projects.getShell(ProjectId.make(row.project_id));
366 if (Option.isNone(project)) return yield* new ArchiveError({ reason: "changed" });
367 return yield* client.publish({
368 environmentId: yield* identity.getEnvironmentId,
369 manifestHash,
370 project: project.value,
371 });
372 }, Effect.mapError(archiveError));
373 const offload = Effect.fn("TranscriptArchive.offload")(function* (threadId: ThreadId) {
374 if (!(yield* client.enabled) || (yield* catalog(threadId)).length > 0) return;
375 const location = yield* client.location;
376 const snapshot = yield* read(
377 threadId,
378 sql
379 .withTransaction(
380 Effect.gen(function* () {
381 const thread = (yield* sql<{
382 archived_at: string | null;
383 deleted_at: string | null;
384 }>`SELECT archived_at, deleted_at FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}`)[0];
385 if (!thread?.archived_at || thread.deleted_at) return undefined;
386 const busy =
387 yield* sql`SELECT 1 FROM (${activeThreads}) WHERE thread_id = ${threadId} LIMIT 1`;
388 if (busy.length) return undefined;
389 const bounds = yield* sql<{
390 sequence: number;
391 }>`SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events WHERE stream_id = ${threadId} AND aggregate_kind = 'thread' AND application_event_version = 2`;
392 const segments = [];
393 for (const table of payloadTables) {
394 const rows = yield* sql<
395 typeof Row.Type
396 >`SELECT * FROM ${sql(`orchestration_v2_projection_${table}`)} WHERE thread_id = ${threadId} ORDER BY rowid`;
397 segments.push({ table, rows });
398 }
399 const events = yield* sql<typeof Row.Type>`SELECT *,
400 CASE WHEN event_type IN ('subagent.updated', 'turn-item.updated') AND json_type(payload_json, '$.promptSequence') = 'integer'
401 THEN json_set(json_remove(payload_json, '$.promptSequence'), '$.prompt', (SELECT json_extract(a.payload_json, '$.prompt') FROM orchestration_events a WHERE a.sequence = json_extract(orchestration_events.payload_json, '$.promptSequence')))
402 ELSE payload_json END AS payload_json
403 FROM orchestration_events WHERE stream_id = ${threadId} AND aggregate_kind = 'thread' AND application_event_version = 2 ORDER BY sequence`;
404 const ancestors = yield* dependencies(threadId);
405 const sourceIds = ancestors.map((row) => row.thread_id);
406 const context: Array<{ table: string; rows: Array<typeof Row.Type> }> = [];
407 for (const name of canonicalTables) {
408 const predicate =
409 name === "context_transfers"
410 ? sql`source_thread_id IN ${sql.in(sourceIds)} OR target_thread_id IN ${sql.in(sourceIds)}`
411 : name === "provider_sessions"
412 ? sql`thread_id IN ${sql.in(sourceIds)} OR provider_session_id IN (SELECT provider_session_id FROM orchestration_v2_projection_provider_session_bindings WHERE thread_id IN ${sql.in(sourceIds)})`
413 : name === "provider_threads"
414 ? sql`thread_id IN ${sql.in(sourceIds)} OR owner_node_id IN (SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id IN ${sql.in(sourceIds)}) OR provider_thread_id IN (SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id IN ${sql.in(sourceIds)})`
415 : sql`thread_id IN ${sql.in(sourceIds)}`;
416 context.push({
417 table: `orchestration_v2_projection_${name}`,
418 rows: [
419 ...(yield* sql<
420 typeof Row.Type
421 >`SELECT * FROM ${sql(`orchestration_v2_projection_${name}`)} WHERE (${predicate}) ${payloadTables.some((table) => table === name) ? sql`AND thread_id <> ${threadId}` : sql``} ORDER BY rowid`),
422 ],
423 });
424 }
425 const revisions = yield* sql<{
426 thread_id: string;
427 sequence: number;
428 }>`SELECT stream_id AS thread_id, MAX(sequence) AS sequence FROM orchestration_events WHERE aggregate_kind = 'thread' AND application_event_version = 2 AND stream_id IN ${sql.in(sourceIds)} GROUP BY stream_id`;
429 return {
430 segments,
431 events,
432 context,
433 revisions,
434 throughSequence: bounds[0]?.sequence ?? 0,
435 };
436 }),
437 )
438 .pipe(NodeSqliteClient.readOnly),
439 );
440 if (!snapshot) return;
441 const files = new Set<string>();
442 const addAttachment = (attachmentId: string) =>
443 Effect.gen(function* () {
444 const input = { attachmentsDir: config.attachmentsDir, attachmentId };
445 const local = resolveAttachmentPathById(input);
446 if (local) {
447 files.add(local);
448 return;
449 }
450 const candidates = attachmentPathsById(input);
451 const archived = (yield* sql<{
452 path: string;
453 }>`SELECT path FROM transcript_archive_assets WHERE path IN ${sql.in(candidates)} AND location = ${location} LIMIT 1`)[0];
454 if (!archived) return yield* new ArchiveError({ reason: "unavailable" });
455 files.add(archived.path);
456 });
457 const workspaces = yield* sql<{
458 thread_id: string;
459 cwd: string | null;
460 }>`SELECT t.thread_id, COALESCE(json_extract(t.payload_json, '$.worktreePath'), p.workspace_root) AS cwd FROM orchestration_v2_projection_threads t LEFT JOIN projection_projects p ON p.project_id = t.project_id WHERE t.thread_id IN ${sql.in(snapshot.revisions.map((row) => row.thread_id))}`;
461 const roots = new Map(workspaces.map((row) => [row.thread_id, row.cwd ?? undefined]));
462 const assetSegments = [
463 ...snapshot.segments.map((segment) => ({
464 ...segment,
465 table: `orchestration_v2_projection_${segment.table}`,
466 })),
467 ...snapshot.context.filter((segment) =>
468 ["orchestration_v2_projection_messages", "orchestration_v2_projection_turn_items"].includes(
469 segment.table,
470 ),
471 ),
472 ];
473 for (const segment of assetSegments)
474 for (const row of segment.rows) {
475 const owner = ThreadId.make(String(row.thread_id));
476 const cwd = roots.get(owner);
477 const payload = decodeRecord(String(row.payload_json));
478 if (
479 segment.table === "orchestration_v2_projection_messages" &&
480 Array.isArray(payload.attachments)
481 )
482 for (const attachment of payload.attachments) {
483 if (
484 attachment &&
485 typeof attachment === "object" &&
486 "id" in attachment &&
487 typeof attachment.id === "string"
488 ) {
489 yield* addAttachment(attachment.id);
490 }
491 }
492 if (segment.table === "orchestration_v2_projection_turn_items")
493 for (const id of threadHtmlRenderAttachmentIds(owner, [
494 {
495 toolName: typeof payload.toolName === "string" ? payload.toolName : null,
496 output: payload.output,
497 },
498 ])) {
499 yield* addAttachment(id);
500 }
501 for (const text of Object.values(payload))
502 if (typeof text === "string")
503 for (const href of extractMarkdownLinkHrefs(text)) {
504 const target = resolveMarkdownFileLinkTarget(href, cwd);
505 if (
506 target &&
507 (isWorkspaceImagePreviewPath(target) ||
508 audioMimeTypeFromExtension(path.extname(target)) ||
509 isWorkspaceVideoPreviewPath(target))
510 )
511 files.add(target);
512 }
513 }
514 const assets = yield* Effect.forEach([...files], (file) =>
515 Effect.gen(function* () {
516 if (yield* fs.exists(file)) return { path: file, hash: (yield* client.upload(file)).hash };
517 const previous = (yield* sql<{
518 location: string;
519 object_hash: string;
520 }>`SELECT location, object_hash FROM transcript_archive_assets WHERE path = ${file} LIMIT 1`)[0];
521 if (!previous || previous.location !== location)
522 return yield* new ArchiveError({ reason: "unavailable" });
523 return { path: file, hash: previous.object_hash };
524 }),
525 );
526 const bundle = {
527 version: 1 as const,
528 threadId,
529 throughSequence: snapshot.throughSequence,
530 segments: snapshot.segments,
531 events: snapshot.events,
532 context: snapshot.context,
533 assets,
534 };
535 const receipt = yield* Effect.scoped(
536 Effect.gen(function* () {
537 const temp = yield* fs.makeTempDirectoryScoped({ prefix: "knife-archive-write-" });
538 let ordinal = 0;
539 const storeRows = (rows: ReadonlyArray<typeof Row.Type>) =>
540 Effect.gen(function* () {
541 const file = path.join(temp, `rows-${ordinal++}`);
542 yield* fs.writeFileString(file, yield* encodeRows(rows), { mode: 0o600 });
543 return (yield* client.upload(file)).hash;
544 });
545 const segments = [];
546 for (const segment of bundle.segments)
547 if (segment.rows.length)
548 segments.push({ table: segment.table, hash: yield* storeRows(segment.rows) });
549 const context = [];
550 for (const segment of bundle.context) {
551 for (const rows of Map.groupBy(
552 segment.rows,
553 (row) => row.thread_id ?? row.source_thread_id ?? "shared",
554 ).values()) {
555 context.push({ table: segment.table, hash: yield* storeRows(rows) });
556 }
557 }
558 const manifest = { ...bundle, segments, context, events: yield* storeRows(bundle.events) };
559 const file = path.join(temp, "transcript");
560 yield* fs.writeFileString(file, yield* encodeBundle(manifest), { mode: 0o600 });
561 const uploaded = yield* client.upload(file);
562 yield* client.download(location, uploaded.hash, path.join(temp, "verified"));
563 return uploaded;
564 }),
565 );
566 const environmentId = yield* identity.getEnvironmentId;
567 const published = yield* publish(threadId, receipt.hash);
568 if (
569 published.environmentId !== environmentId ||
570 published.thread.id !== threadId ||
571 published.manifestHash !== receipt.hash ||
572 published.throughSequence !== snapshot.throughSequence
573 )
574 return yield* new ArchiveError({ reason: "integrity" });
575 yield* sql`INSERT INTO transcript_archive_publications(thread_id, location, object_hash) VALUES(${threadId}, ${location}, ${receipt.hash}) ON CONFLICT(thread_id) DO UPDATE SET location = excluded.location, object_hash = excluded.object_hash`;
576 yield* sql.withTransaction(
577 Effect.gen(function* () {
578 const rows =
579 yield* sql`SELECT 1 FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId} AND archived_at IS NOT NULL AND deleted_at IS NULL AND NOT EXISTS (SELECT 1 FROM transcript_archives WHERE thread_id = ${threadId}) AND NOT EXISTS (SELECT 1 FROM (${activeThreads}) WHERE thread_id = ${threadId}) AND ${snapshot.throughSequence} = (SELECT MAX(sequence) FROM orchestration_events WHERE stream_id = ${threadId} AND aggregate_kind = 'thread' AND application_event_version = 2)`;
580 if (!rows.length) return yield* new ArchiveError({ reason: "changed" });
581 for (const revision of snapshot.revisions) {
582 const changed =
583 yield* sql`SELECT 1 WHERE ${revision.sequence} <> (SELECT MAX(sequence) FROM orchestration_events WHERE aggregate_kind = 'thread' AND application_event_version = 2 AND stream_id = ${revision.thread_id})`;
584 if (changed.length) return yield* new ArchiveError({ reason: "changed" });
585 }
586 const metadata: typeof Metadata.Type = snapshot.segments.map((segment) => ({
587 ...segment,
588 rows: segment.rows.map((row) => ({
589 ...row,
590 payload_json: metadataPayload(segment.table, String(row.payload_json)),
591 })),
592 }));
593 yield* sql`INSERT INTO transcript_archives(thread_id, location, object_hash, through_sequence, metadata_json) VALUES(${threadId}, ${location}, ${receipt.hash}, ${snapshot.throughSequence}, ${encodeMetadata(metadata)})`;
594 for (const segment of metadata)
595 for (const row of segment.rows)
596 yield* sql`UPDATE ${sql(`orchestration_v2_projection_${segment.table}`)} SET payload_json = ${row.payload_json} WHERE ${sql(ids[segment.table])} = ${row[ids[segment.table]]} AND thread_id = ${threadId}`;
597 yield* sql`UPDATE orchestration_events SET payload_json = json_set(json_remove(payload_json, '$.promptSequence'), '$.prompt', (SELECT json_extract(a.payload_json, '$.prompt') FROM orchestration_events a WHERE a.sequence = json_extract(orchestration_events.payload_json, '$.promptSequence'))) WHERE json_type(payload_json, '$.promptSequence') = 'integer' AND json_extract(payload_json, '$.promptSequence') IN (SELECT sequence FROM orchestration_events WHERE stream_id = ${threadId})`;
598 yield* sql`UPDATE orchestration_events SET payload_json = 'null' WHERE stream_id = ${threadId} AND application_event_version = 2 AND aggregate_kind = 'thread' AND sequence <= ${snapshot.throughSequence} AND event_type IN ('message.updated', 'turn-item.updated', 'subagent.updated')`;
599 for (const asset of assets)
600 yield* sql`INSERT INTO transcript_archive_assets(thread_id, path, location, object_hash) VALUES(${threadId}, ${asset.path}, ${location}, ${asset.hash}) ON CONFLICT(thread_id, path) DO UPDATE SET location = excluded.location, object_hash = excluded.object_hash`;
601 yield* sql`DELETE FROM transcript_archive_failures WHERE thread_id = ${threadId}`;
602 yield* sql`UPDATE transcript_archive_reclamation SET pending = 1 WHERE singleton = 1`;
603 }),
604 );
605 }, Effect.mapError(archiveError));
606 const evictAttachments = sql.withTransaction(
607 Effect.gen(function* () {
608 const assets = yield* sql<{
609 thread_id: ThreadId;
610 path: string;
611 }>`SELECT a.thread_id, a.path FROM transcript_archive_assets a JOIN transcript_archives c ON c.thread_id = a.thread_id`;
612 const references = yield* sql<{
613 id: string;
614 }>`SELECT DISTINCT json_extract(a.value, '$.id') AS id FROM orchestration_v2_projection_messages m JOIN orchestration_v2_projection_threads t ON t.thread_id = m.thread_id, json_each(m.payload_json, '$.attachments') a WHERE t.archived_at IS NULL AND t.deleted_at IS NULL`;
615 const warmSources = yield* sql<{ thread_id: string }>`WITH RECURSIVE sources(thread_id) AS (
616 SELECT thread_id FROM orchestration_v2_projection_threads WHERE archived_at IS NULL AND deleted_at IS NULL
617 UNION SELECT json_extract(t.payload_json, '$.forkedFrom.threadId') FROM orchestration_v2_projection_threads t JOIN sources s ON s.thread_id = t.thread_id WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run'
618 ) SELECT thread_id FROM sources`;
619 const referenced = new Set(references.map((row) => row.id));
620 const warm = new Set(warmSources.map((row) => row.thread_id));
621 for (const asset of assets) {
622 const relative = path.relative(config.attachmentsDir, asset.path);
623 if (relative.startsWith("..") || path.isAbsolute(relative) || warm.has(asset.thread_id))
624 continue;
625 const id = parseAttachmentIdFromRelativePath(relative);
626 if (
627 !id ||
628 referenced.has(id) ||
629 parseThreadSegmentFromAttachmentId(id) !== toSafeThreadAttachmentSegment(asset.thread_id)
630 )
631 continue;
632 yield* fs.remove(asset.path, { force: true });
633 }
634 }),
635 );
636 const sweep = Effect.gen(function* () {
637 if (!(yield* client.enabled)) return;
638 const environmentId = yield* identity.getEnvironmentId;
639 const now = yield* Clock.currentTimeMillis;
640 const attempt = (threadId: ThreadId, operation: Effect.Effect<void, ArchiveError>) =>
641 operation.pipe(
642 Effect.andThen(sql`DELETE FROM transcript_archive_failures WHERE thread_id = ${threadId}`),
643 Effect.catchTags({
644 ArchiveError: (error) =>
645 error.reason === "changed"
646 ? Effect.void
647 : sql`INSERT INTO transcript_archive_failures(thread_id, attempts, reason, next_attempt_at) VALUES(${threadId}, 1, ${error.reason}, ${now + 60_000}) ON CONFLICT(thread_id) DO UPDATE SET attempts = attempts + 1, reason = excluded.reason, next_attempt_at = ${now} + MIN(3600000, 60000 * (1 << MIN(attempts, 6)))`.pipe(
648 Effect.andThen(
649 Effect.logWarning("Transcript archive sync will retry", {
650 threadId,
651 reason: error.reason,
652 }),
653 ),
654 ),
655 }),
656 );
657 const unpublished = yield* sql<{
658 thread_id: ThreadId;
659 object_hash: string;
660 location: string;
661 }>`SELECT a.thread_id, a.object_hash, a.location FROM transcript_archives a LEFT JOIN transcript_archive_publications p ON p.thread_id = a.thread_id LEFT JOIN transcript_archive_failures f ON f.thread_id = a.thread_id WHERE p.thread_id IS NULL AND COALESCE(f.next_attempt_at, 0) <= ${now} LIMIT 10`;
662 for (const entry of unpublished) {
663 if (entry.location !== (yield* client.location)) continue;
664 yield* attempt(
665 entry.thread_id,
666 Effect.gen(function* () {
667 yield* publish(entry.thread_id, entry.object_hash);
668 yield* sql`INSERT INTO transcript_archive_publications(thread_id, location, object_hash) VALUES(${entry.thread_id}, ${entry.location}, ${entry.object_hash}) ON CONFLICT(thread_id) DO UPDATE SET location = excluded.location, object_hash = excluded.object_hash`;
669 }).pipe(Effect.mapError(archiveError)),
670 );
671 }
672 const warm = yield* sql<{
673 thread_id: ThreadId;
674 location: string;
675 object_hash: string;
676 }>`SELECT p.thread_id, p.location, p.object_hash FROM transcript_archive_publications p JOIN orchestration_v2_projection_threads t ON t.thread_id = p.thread_id LEFT JOIN transcript_archive_failures f ON f.thread_id = p.thread_id WHERE (t.archived_at IS NULL OR t.deleted_at IS NOT NULL) AND COALESCE(f.next_attempt_at, 0) <= ${now}`;
677 for (const entry of warm) {
678 if (entry.location !== (yield* client.location)) continue;
679 yield* attempt(
680 entry.thread_id,
681 Effect.gen(function* () {
682 yield* client.withdraw(environmentId, entry.thread_id, entry.object_hash);
683 yield* sql`DELETE FROM transcript_archive_publications WHERE thread_id = ${entry.thread_id} AND object_hash = ${entry.object_hash}`;
684 }).pipe(Effect.mapError(archiveError)),
685 );
686 }
687 const candidates = yield* sql<{
688 thread_id: ThreadId;
689 }>`SELECT t.thread_id FROM orchestration_v2_projection_threads t LEFT JOIN transcript_archive_failures f ON f.thread_id = t.thread_id WHERE t.archived_at IS NOT NULL AND t.deleted_at IS NULL AND COALESCE(f.next_attempt_at, 0) <= ${now} AND NOT EXISTS (SELECT 1 FROM transcript_archives a WHERE a.thread_id = t.thread_id) AND NOT EXISTS (SELECT 1 FROM (${activeThreads}) active WHERE active.thread_id = t.thread_id) ORDER BY t.archived_at LIMIT 10`;
690 for (const row of candidates) yield* attempt(row.thread_id, offload(row.thread_id));
691 yield* evictAttachments;
692 yield* storage.reclaim;
693 }).pipe(Effect.mapError(archiveError));
694 const queuedThreads = sql`SELECT t.thread_id FROM orchestration_v2_projection_threads t LEFT JOIN transcript_archives a ON a.thread_id = t.thread_id LEFT JOIN transcript_archive_publications p ON p.thread_id = t.thread_id WHERE (t.archived_at IS NOT NULL AND t.deleted_at IS NULL AND (a.thread_id IS NULL OR p.thread_id IS NULL)) OR ((t.archived_at IS NULL OR t.deleted_at IS NOT NULL) AND p.thread_id IS NOT NULL)`;
695 const manage = Effect.fn("TranscriptArchive.manage")(function* (
696 request: ArchiveManagementRequest,
697 ) {
698 if (request.type === "archive.retry")
699 yield* sql`DELETE FROM transcript_archive_failures ${request.threadId ? sql`WHERE thread_id = ${request.threadId}` : sql``}`;
700 const counts = yield* sql<{
701 offloaded: number;
702 pending: number;
703 }>`SELECT (SELECT COUNT(*) FROM transcript_archives) AS offloaded, (SELECT COUNT(*) FROM (${queuedThreads})) AS pending`;
704 const failures = yield* sql<{
705 threadId: ThreadId;
706 attempts: number;
707 reason: string;
708 }>`SELECT thread_id AS threadId, attempts, reason FROM transcript_archive_failures ORDER BY thread_id`;
709 return {
710 enabled: yield* client.enabled,
711 queue: yield* sql<{
712 threadId: ThreadId;
713 nextAttemptAt: number;
714 }>`SELECT t.thread_id AS threadId, COALESCE(f.next_attempt_at, 0) AS nextAttemptAt FROM (${queuedThreads}) t LEFT JOIN transcript_archive_failures f ON f.thread_id = t.thread_id ORDER BY nextAttemptAt, t.thread_id`,
715 storage: yield* storage.inspect,
716 lastReclamation:
717 (yield* sql<{
718 beforeAllocatedBytes: number;
719 afterAllocatedBytes: number;
720 }>`SELECT before_allocated_bytes AS beforeAllocatedBytes, after_allocated_bytes AS afterAllocatedBytes FROM transcript_archive_reclamation WHERE singleton = 1 AND completed_at IS NOT NULL`)[0] ??
721 null,
722 offloaded: counts[0]?.offloaded ?? 0,
723 pending: counts[0]?.pending ?? 0,
724 failures,
725 };
726 }, Effect.mapError(archiveError));
727 return TranscriptArchive.of({
728 enabled: client.enabled,
729 manage,
730 offload,
731 restore,
732 read,
733 readStream: (id, stream) =>
734 Stream.unwrap(
735 entries(id).pipe(
736 Effect.mapError(archiveError),
737 Effect.flatMap((found) =>
738 found.length === 0
739 ? Effect.succeed(stream)
740 : read(id, Stream.runCollect(stream)).pipe(Effect.map(Stream.fromIterable)),
741 ),
742 ),
743 ),
744 sweep,
745 asset: (paths) =>
746 sql<{
747 path: string;
748 location: string;
749 object_hash: string;
750 }>`SELECT path, location, object_hash FROM transcript_archive_assets WHERE path IN ${sql.in(paths)} ORDER BY rowid DESC LIMIT 1`.pipe(
751 Effect.map((rows) =>
752 Option.fromUndefinedOr(rows[0]).pipe(
753 Option.map((row) => ({
754 path: row.path,
755 location: row.location,
756 hash: row.object_hash,
757 })),
758 ),
759 ),
760 Effect.mapError(archiveError),
761 ),
762 });
763});
764
765export const layer = Layer.effect(TranscriptArchive, make).pipe(Layer.provide(ProjectStore.layer));
766
767/** Rebuild cold timeline indexes locally; no NAS request or transcript body is needed. */
768export const rebuildMetadata = Effect.gen(function* () {
769 const sql = yield* SqlClient.SqlClient;
770 const rows = yield* sql<{
771 metadata_json: string;
772 }>`SELECT metadata_json FROM transcript_archives ORDER BY through_sequence`;
773 for (const entry of rows)
774 for (const segment of yield* decodeMetadata(entry.metadata_json)) {
775 for (const row of segment.rows)
776 yield* sql`INSERT INTO ${sql(`orchestration_v2_projection_${segment.table}`)} ${sql.insert(row)} ON CONFLICT DO NOTHING`;
777 }
778});