1import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, it } from "@effect/vitest";
4import {
5 AssetCreateUrlResult,
6 ArchiveCatalogResult,
7 AuthAdministrativeScopes,
8 AuthSessionId,
9 HubError,
10 ORCHESTRATION_PROTOCOL_VERSION,
11 OrchestrationV2ThreadDetailSnapshot,
12 AuthArchiveReadScope,
13 AuthArchiveWriteScope,
14 AuthOrchestrationReadScope,
15 ArchiveError,
16 EnvironmentId,
17 EventId,
18 MessageId,
19 ProjectId,
20 ProviderInstanceId,
21 ThreadId,
22 TurnItemId,
23 ChatAttachmentId,
24 RunId,
25 NodeId,
26 ProviderDriverKind,
27 type OrchestrationV2AppThread,
28} from "@t3tools/contracts";
29import * as TestClock from "effect/testing/TestClock";
30import * as ArchiveCatalog from "./ArchiveCatalog.ts";
31import * as Hub from "../environment/Hub.ts";
32import * as HubInbox from "../environment/HubInbox.ts";
33import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
34import * as HostedAuth from "../auth/HostedAuth.ts";
35import * as HostedAuthConfig from "../auth/HostedAuthConfig.ts";
36import * as SessionStore from "../auth/SessionStore.ts";
37import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
38import * as ArchiveAssets from "./ArchiveAssets.ts";
39import * as ArchiveHttp from "./ArchiveHttp.ts";
40import * as Option from "effect/Option";
41import * as Redacted from "effect/Redacted";
42import * as Schema from "effect/Schema";
43import * as ConfigProvider from "effect/ConfigProvider";
44import * as DateTime from "effect/DateTime";
45import * as Deferred from "effect/Deferred";
46import * as Fiber from "effect/Fiber";
47import * as HostProcess from "@t3tools/shared/HostProcess";
48import * as ArchiveConnection from "./ArchiveConnection.ts";
49import * as SharedServer from "../environment/SharedServer.ts";
50import * as ServerSettings from "../serverSettings.ts";
51import * as ArchiveClient from "./ArchiveClient.ts";
52import * as Effect from "effect/Effect";
53import * as FileSystem from "effect/FileSystem";
54import * as Layer from "effect/Layer";
55import * as Path from "effect/Path";
56import * as PlatformError from "effect/PlatformError";
57import * as Crypto from "effect/Crypto";
58import * as SqlClient from "effect/sql/SqlClient";
59import * as Stream from "effect/Stream";
60import { HttpClient, HttpRouter } from "effect/http";
61import * as ServerConfig from "../config.ts";
62import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
63import * as SqlitePersistence from "../persistence/Sqlite.ts";
64import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
65import * as ProjectionMaintenance from "../orchestration-v2/ProjectionMaintenance.ts";
66import * as EventStore from "../orchestration-v2/EventStore.ts";
67import * as EventSink from "../orchestration-v2/EventSink.ts";
68import * as ApplicationEvents from "../persistence/OrchestrationEventStore.ts";
69import * as ArchivedProjectionStore from "./ArchivedProjectionStore.ts";
70import * as ArchiveObjects from "./ArchiveObjects.ts";
71import * as ArchiveRuntime from "./runtimeLayer.ts";
72import * as TranscriptArchive from "./TranscriptArchive.ts";
73import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
74import { createDeterministicAttachmentId } from "../attachmentStore.ts";
75import { ASSET_ROUTE_PREFIX, issueAssetUrl, resolveAsset } from "../assets/AssetAccess.ts";
76import * as NativeAppIconResolver from "../assets/NativeAppIconResolver.ts";
77import * as ProjectFaviconResolver from "../project/ProjectFaviconResolver.ts";
78import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
79
80const decodeAssetUrl = Schema.decodeUnknownEffect(AssetCreateUrlResult);
81const decodeCatalog = Schema.decodeUnknownEffect(Schema.toCodecJson(ArchiveCatalogResult));
82const decodeSnapshot = Schema.decodeUnknownEffect(
83 Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot),
84);
85const threadId = ThreadId.make("archive-review");
86const fixture = (physical = false) =>
87 Effect.gen(function* () {
88 const fs = yield* FileSystem.FileSystem;
89 const path = yield* Path.Path;
90 const temporary = yield* fs.makeTempDirectoryScoped({ prefix: "knife-archive-test-" });
91 const root = path.join(temporary, "nas");
92 yield* fs.makeDirectory(root);
93 yield* fs.writeFileString(path.join(root, ".knife-archive-store-id"), "archive-test-store");
94 const config = yield* ServerConfig.ServerConfig.pipe(
95 Effect.provide(ServerConfig.layerTest(temporary, { prefix: "knife-archive-test-" })),
96 );
97 const database = physical
98 ? SqlitePersistence.layerFromPath(path.join(temporary, "source.sqlite"))
99 : SqlitePersistence.layerMemory;
100 const runtime = ArchiveRuntime.layer.pipe(
101 Layer.provide(Layer.succeed(HostProcess.HomeDirectory, temporary)),
102 Layer.provide(ArchiveConnection.layerLegacy.pipe(Layer.provide(SharedServer.layer))),
103 Layer.provideMerge(ServerSettings.layerTest()),
104 Layer.provideMerge(ServerSecretStore.layer),
105 Layer.provideMerge(Layer.succeed(ServerConfig.ServerConfig, config)),
106 Layer.provideMerge(database),
107 Layer.provide(
108 Layer.succeed(
109 HttpClient.HttpClient,
110 HttpClient.make(() => Effect.die("Unexpected HTTP request")),
111 ),
112 ),
113 Layer.provide(
114 ConfigProvider.layer(
115 ConfigProvider.fromUnknown({
116 T3CODE_ARCHIVE_DIR: root,
117 T3CODE_ARCHIVE_STORE_ID: "archive-test-store",
118 T3CODE_ARCHIVE_OFFLOAD: "true",
119 }),
120 ),
121 ),
122 );
123 const stores = Layer.mergeAll(
124 ArchivedProjectionStore.layer,
125 EventStore.layerFromOrchestrationEventStore.pipe(Layer.provideMerge(ApplicationEvents.layer)),
126 ).pipe(Layer.provideMerge(runtime));
127 const layer = Layer.mergeAll(
128 stores,
129 EventSink.layer.pipe(Layer.provide(stores)),
130 ProjectionMaintenance.layer.pipe(Layer.provide(ProjectionStore.layer), Layer.provide(stores)),
131 );
132 return { layer, root, path, temporary };
133 });
134
135const seed = Effect.gen(function* () {
136 const sink = yield* EventSink.EventSinkV2;
137 const now = yield* DateTime.now;
138 const instanceId = ProviderInstanceId.make("codex");
139 yield* (yield* SqlClient.SqlClient)`INSERT OR IGNORE INTO projection_projects(project_id, title, workspace_root, scripts_json, created_at, updated_at) VALUES(${ProjectId.make("archive-project")}, 'Archive workspace', '/archive/workspace', '[]', ${DateTime.formatIso(now)}, ${DateTime.formatIso(now)})`;
140 const thread: OrchestrationV2AppThread = {
141 createdBy: "user",
142 creationSource: "web",
143 id: threadId,
144 projectId: ProjectId.make("archive-project"),
145 title: "Searchable archive title",
146 providerInstanceId: instanceId,
147 modelSelection: { instanceId, model: "model" },
148 runtimeMode: "full-access",
149 interactionMode: "default",
150 branch: null,
151 worktreePath: null,
152 activeProviderThreadId: null,
153 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
154 forkedFrom: null,
155 createdAt: now,
156 updatedAt: now,
157 archivedAt: null,
158 settledOverride: null,
159 settledAt: null,
160 lastVisitedAt: null,
161 deletedAt: null,
162 };
163 yield* sink.write({
164 events: [
165 {
166 id: EventId.make("archive-created"),
167 type: "thread.created",
168 threadId,
169 occurredAt: now,
170 payload: thread,
171 },
172 {
173 id: EventId.make("archive-message"),
174 type: "message.updated",
175 threadId,
176 occurredAt: now,
177 payload: {
178 id: MessageId.make("archive-message"),
179 threadId,
180 runId: null,
181 nodeId: null,
182 role: "assistant",
183 createdBy: "agent",
184 creationSource: "provider",
185 text: "private archived message body",
186 attachments: [],
187 streaming: false,
188 createdAt: now,
189 updatedAt: now,
190 },
191 },
192 {
193 id: EventId.make("archive-item"),
194 type: "turn-item.updated",
195 threadId,
196 occurredAt: now,
197 payload: {
198 id: TurnItemId.make("archive-item"),
199 threadId,
200 runId: null,
201 nodeId: null,
202 type: "assistant_message",
203 providerThreadId: null,
204 providerTurnId: null,
205 nativeItemRef: null,
206 parentItemId: null,
207 title: null,
208 startedAt: now,
209 completedAt: now,
210 streaming: false,
211 status: "completed",
212 ordinal: 1,
213 messageId: MessageId.make("archive-message"),
214 text: "private archived message body",
215 updatedAt: now,
216 },
217 },
218 {
219 id: EventId.make("archive-archived"),
220 type: "thread.archived",
221 threadId,
222 occurredAt: now,
223 payload: { ...thread, archivedAt: now },
224 },
225 ],
226 });
227 return thread;
228});
229
230it.layer(NodeServices.layer)("transcript archive", (it) => {
231 it.effect("reads the durable catalog after the source database is closed", () =>
232 Effect.gen(function* () {
233 const f = yield* fixture();
234 const environmentId = yield* Effect.gen(function* () {
235 yield* seed;
236 yield* (yield* TranscriptArchive.TranscriptArchive).offload(threadId);
237 return yield* (yield* ServerEnvironment.ServerEnvironmentIdentity).getEnvironmentId;
238 }).pipe(Effect.provide(f.layer));
239 const central = ArchiveCatalog.layer.pipe(
240 Layer.provideMerge(ArchiveObjects.layer),
241 Layer.provide(
242 ConfigProvider.layer(ConfigProvider.fromUnknown({ T3CODE_ARCHIVE_DIR: f.root })),
243 ),
244 );
245 yield* Effect.gen(function* () {
246 const catalog = yield* ArchiveCatalog.ArchiveCatalog;
247 const entries = (yield* catalog.list).entries;
248 assert.strictEqual(entries.length, 1);
249 assert.strictEqual(entries[0]?.environmentId, environmentId);
250 const snapshot = yield* catalog.read(environmentId, threadId);
251 assert.strictEqual(snapshot.projection.messages[0]?.text, "private archived message body");
252 assert.strictEqual(
253 snapshot.projection.turnItems.find((item) => item.type === "assistant_message")?.text,
254 "private archived message body",
255 );
256 assert.isNotNull(snapshot.projection.thread.archivedAt);
257 const descriptor = {
258 environmentId,
259 label: "Sleeping laptop",
260 platform: { os: "linux" as const, arch: "x64" as const },
261 serverVersion: "test",
262 orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
263 capabilities: { repositoryIdentity: false },
264 };
265 const hub = Hub.Hub.of({
266 directory: () =>
267 Effect.succeed({
268 hubEnvironmentId: EnvironmentId.make("central-nas"),
269 environments: [
270 {
271 environmentId,
272 label: descriptor.label,
273 httpBaseUrl: "https://archive.example.test/environments/source",
274 scopes: AuthAdministrativeScopes,
275 descriptor,
276 },
277 ],
278 }),
279 authorize: () => Effect.succeed(AuthAdministrativeScopes),
280 connect: () => Effect.fail(new HubError({ reason: "upstream" })),
281 descriptor: () => Effect.succeed(descriptor),
282 manage: () => Effect.die("Unexpected management"),
283 });
284 const discovered = yield* Effect.gen(function* () {
285 const inbox = yield* HubInbox.HubInbox;
286 const sql = yield* SqlClient.SqlClient;
287 assert.strictEqual((yield* sql`SELECT 1 FROM hub_inbox`).length, 0);
288 return yield* inbox.read({
289 sessionId: AuthSessionId.make("fresh-desktop"),
290 subject: "owner",
291 method: "bearer-access-token",
292 scopes: AuthAdministrativeScopes,
293 });
294 }).pipe(
295 Effect.provide(
296 HubInbox.layer.pipe(
297 Layer.provide(Layer.succeed(Hub.Hub, hub)),
298 Layer.provide(Layer.succeed(ArchiveCatalog.ArchiveCatalog, catalog)),
299 Layer.provide(
300 Layer.succeed(
301 HttpClient.HttpClient,
302 HttpClient.make(() => Effect.die("Source must not be contacted")),
303 ),
304 ),
305 Layer.provideMerge(Layer.fresh(SqlitePersistence.layerMemory)),
306 ),
307 ),
308 );
309 assert.strictEqual(discovered.environments[0]?.environmentId, environmentId);
310 assert.strictEqual(discovered.environments[0]?.snapshot.archivedThreads[0]?.id, threadId);
311 assert.strictEqual(
312 discovered.environments[0]?.snapshot.projects[0]?.id,
313 ProjectId.make("archive-project"),
314 );
315 assert.strictEqual(
316 discovered.environments[0]?.snapshot.projects[0]?.workspaceRoot,
317 "/archive/workspace",
318 );
319 assert.strictEqual(
320 (yield* catalog.read(EnvironmentId.make("other-source"), threadId).pipe(Effect.flip))
321 .reason,
322 "unavailable",
323 );
324 const fs = yield* FileSystem.FileSystem;
325 yield* fs.writeFileString(
326 f.path.join(f.root, "objects", entries[0]!.manifestHash),
327 "corrupt",
328 );
329 assert.strictEqual(
330 (yield* catalog.read(environmentId, threadId).pipe(Effect.flip)).reason,
331 "integrity",
332 );
333 }).pipe(Effect.provide(central));
334 }),
335 );
336
337 it.effect(
338 "keeps source payloads when publication cannot verify the complete object closure",
339 () =>
340 Effect.gen(function* () {
341 const f = yield* fixture();
342 yield* Effect.gen(function* () {
343 yield* seed;
344 const client = yield* ArchiveClient.ArchiveClient;
345 const fs = yield* FileSystem.FileSystem;
346 const archive = yield* TranscriptArchive.TranscriptArchive.pipe(
347 Effect.provide(Layer.fresh(TranscriptArchive.layer)),
348 Effect.provideService(ArchiveClient.ArchiveClient, {
349 ...client,
350 publish: (input) =>
351 Effect.gen(function* () {
352 const bundle = JSON.parse(
353 yield* fs.readFileString(f.path.join(f.root, "objects", input.manifestHash)),
354 );
355 yield* fs.remove(f.path.join(f.root, "objects", bundle.events));
356 return yield* client.publish(input);
357 }),
358 }),
359 );
360 assert.strictEqual(
361 (yield* archive.offload(threadId).pipe(Effect.flip)).reason,
362 "storage",
363 );
364 assert.strictEqual(
365 (yield* (yield* SqlClient.SqlClient)`SELECT 1 FROM transcript_archives`).length,
366 0,
367 );
368 assert.strictEqual(
369 (yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadProjection(threadId))
370 .messages[0]?.text,
371 "private archived message body",
372 );
373 assert.strictEqual(
374 (yield* (yield* ArchiveCatalog.ArchiveCatalog).list).entries.length,
375 0,
376 );
377 }).pipe(Effect.provide(f.layer));
378 }),
379 );
380
381 it.effect("persists failed uploads across restart and retries after their backoff", () =>
382 Effect.gen(function* () {
383 const f = yield* fixture(true);
384 const expectedEnvironmentId = yield* Effect.gen(function* () {
385 yield* seed;
386 const client = yield* ArchiveClient.ArchiveClient;
387 const archive = yield* TranscriptArchive.TranscriptArchive.pipe(
388 Effect.provide(Layer.fresh(TranscriptArchive.layer)),
389 Effect.provideService(ArchiveClient.ArchiveClient, {
390 ...client,
391 upload: () => Effect.fail(new ArchiveError({ reason: "unavailable" })),
392 }),
393 );
394 yield* archive.sweep;
395 const status = yield* archive.manage({ type: "archive.status" });
396 assert.strictEqual(status.failures[0]?.attempts, 1);
397 assert.strictEqual(status.offloaded, 0);
398 return yield* (yield* ServerEnvironment.ServerEnvironmentIdentity).getEnvironmentId;
399 }).pipe(Effect.provide(f.layer));
400 yield* Effect.gen(function* () {
401 const archive = yield* TranscriptArchive.TranscriptArchive;
402 assert.strictEqual(
403 yield* (yield* ServerEnvironment.ServerEnvironmentIdentity).getEnvironmentId,
404 expectedEnvironmentId,
405 );
406 yield* archive.sweep;
407 assert.strictEqual(
408 (yield* archive.manage({ type: "archive.status" })).failures[0]?.attempts,
409 1,
410 );
411 yield* TestClock.adjust("1 minute");
412 yield* archive.sweep;
413 assert.strictEqual((yield* archive.manage({ type: "archive.status" })).offloaded, 1);
414 }).pipe(Effect.provide(Layer.fresh(f.layer)));
415 }),
416 );
417
418 it.effect(
419 "reclaims allocated SQLite payload bytes and retains measured native provider logs",
420 () =>
421 Effect.gen(function* () {
422 const f = yield* fixture(true);
423 const body = "large archived body ".repeat(100_000);
424 yield* Effect.gen(function* () {
425 yield* seed;
426 const archive = yield* TranscriptArchive.TranscriptArchive;
427 const projections = yield* ProjectionStore.ProjectionStoreV2;
428 const sql = yield* SqlClient.SqlClient;
429 const fs = yield* FileSystem.FileSystem;
430 const config = yield* ServerConfig.ServerConfig;
431 const message = (yield* projections.getThreadProjection(threadId)).messages[0]!;
432 yield* (yield* EventSink.EventSinkV2).write({
433 events: [
434 {
435 id: EventId.make("large-body"),
436 type: "message.updated",
437 threadId,
438 occurredAt: yield* DateTime.now,
439 payload: { ...message, text: body },
440 },
441 ],
442 });
443 yield* fs.makeDirectory(config.providerLogsDir, { recursive: true });
444 const logFile = f.path.join(config.providerLogsDir, "provider.jsonl");
445 yield* fs.writeFileString(logFile, "native provider log stays local");
446 yield* sql`PRAGMA wal_checkpoint(TRUNCATE)`;
447 const before = yield* archive.manage({ type: "archive.status" });
448 yield* archive.sweep;
449 const after = yield* archive.manage({ type: "archive.status" });
450 assert.isAbove(
451 before.storage.allocatedDatabaseBytes,
452 after.storage.allocatedDatabaseBytes,
453 );
454 assert.isAbove(
455 after.lastReclamation!.beforeAllocatedBytes,
456 after.lastReclamation!.afterAllocatedBytes,
457 );
458 assert.strictEqual(
459 after.storage.nativeProviderLogBytes,
460 Buffer.byteLength("native provider log stays local"),
461 );
462 assert.strictEqual(yield* fs.readFileString(logFile), "native provider log stays local");
463 assert.strictEqual(
464 (yield* projections.getThreadProjection(threadId)).messages[0]?.text,
465 body,
466 );
467 yield* (yield* ProjectionMaintenance.ProjectionMaintenanceV2).rebuild;
468 assert.strictEqual(
469 (yield* projections.getThreadProjection(threadId)).messages[0]?.text,
470 body,
471 );
472 }).pipe(Effect.provide(f.layer));
473 yield* Effect.gen(function* () {
474 const archive = yield* TranscriptArchive.TranscriptArchive;
475 const projections = yield* ProjectionStore.ProjectionStoreV2;
476 assert.strictEqual(
477 (yield* projections.getThreadProjection(threadId)).messages[0]?.text,
478 body,
479 );
480 yield* archive.restore(threadId);
481 const thread = (yield* projections.getThreadProjection(threadId)).thread;
482 yield* (yield* EventSink.EventSinkV2).write({
483 events: [
484 {
485 id: EventId.make("restart-unarchive"),
486 type: "thread.unarchived",
487 threadId,
488 occurredAt: yield* DateTime.now,
489 payload: { ...thread, archivedAt: null },
490 },
491 ],
492 });
493 const restored = yield* projections.getThreadProjection(threadId);
494 assert.isNull(restored.thread.archivedAt);
495 assert.strictEqual(restored.messages[0]?.text, body);
496 }).pipe(Effect.provide(Layer.fresh(f.layer)));
497 }),
498 );
499
500 it.effect(
501 "enforces narrow HTTP archive scopes and the authenticated device environment binding",
502 () =>
503 Effect.gen(function* () {
504 const f = yield* fixture();
505 const settings = {
506 publicUrl: "https://archive.example.test",
507 issuer: "https://identity.example.test",
508 clientId: "archive-test",
509 clientSecret: Redacted.make("test-only"),
510 ownerSub: "test-owner",
511 maxInteractiveAgeSeconds: 3600,
512 issuerRevalidationSeconds: 3600,
513 };
514 const layers = ArchiveAssets.layer.pipe(
515 Layer.provideMerge(NodeHttpPlatform.layer),
516 Layer.provideMerge(
517 HostedAuth.layer.pipe(
518 Layer.provideMerge(EnvironmentAuth.layer),
519 Layer.provide(
520 Layer.succeed(HostedAuthConfig.HostedAuthConfig, Option.some(settings)),
521 ),
522 Layer.provide(
523 Layer.succeed(
524 HttpClient.HttpClient,
525 HttpClient.make(() => Effect.die("Unexpected identity request")),
526 ),
527 ),
528 Layer.provideMerge(f.layer),
529 ),
530 ),
531 );
532 yield* Effect.gen(function* () {
533 yield* seed;
534 const fs = yield* FileSystem.FileSystem;
535 const config = yield* ServerConfig.ServerConfig;
536 const imageId = ChatAttachmentId.make(
537 createDeterministicAttachmentId(threadId, "nas-image")!,
538 );
539 const movieId = ChatAttachmentId.make(
540 `${createDeterministicAttachmentId(threadId, "nas-movie")!}-mp4`,
541 );
542 yield* fs.makeDirectory(config.attachmentsDir, { recursive: true });
543 const imageFile = f.path.join(config.attachmentsDir, `${imageId}.png`);
544 const movieFile = f.path.join(config.attachmentsDir, `${movieId}.mp4`);
545 yield* fs.writeFile(imageFile, new Uint8Array([137, 80, 78, 71, 1, 2, 3]));
546 yield* fs.writeFileString(movieFile, "0123456789");
547 const message = (yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadProjection(
548 threadId,
549 )).messages[0]!;
550 yield* (yield* EventSink.EventSinkV2).write({
551 events: [
552 {
553 id: EventId.make("nas-attachments"),
554 type: "message.updated",
555 threadId,
556 occurredAt: yield* DateTime.now,
557 payload: {
558 ...message,
559 attachments: [
560 {
561 type: "image",
562 id: imageId,
563 name: "picture.png",
564 mimeType: "image/png",
565 sizeBytes: 7,
566 },
567 {
568 type: "file",
569 id: movieId,
570 name: "movie.mp4",
571 mimeType: "video/mp4",
572 sizeBytes: 10,
573 },
574 ],
575 },
576 },
577 ],
578 });
579 yield* (yield* TranscriptArchive.TranscriptArchive).sweep;
580 assert.isFalse(yield* fs.exists(imageFile));
581 assert.isFalse(yield* fs.exists(movieFile));
582 const environmentId = yield* (yield* ServerEnvironment.ServerEnvironmentIdentity)
583 .getEnvironmentId;
584 const auth = yield* EnvironmentAuth.EnvironmentAuth;
585 const read = yield* auth.issueSession({ scopes: [AuthArchiveReadScope] });
586 const write = yield* auth.issueSession({ scopes: [AuthArchiveWriteScope] });
587 const legacy = yield* auth.issueSession({ scopes: [AuthOrchestrationReadScope] });
588 const sql = yield* SqlClient.SqlClient;
589 const accountHash = Buffer.from(
590 yield* (yield* Crypto.Crypto).digest(
591 "SHA-256",
592 new TextEncoder().encode(`${settings.issuer}\0${settings.ownerSub}`),
593 ),
594 ).toString("base64url");
595 const now = yield* DateTime.now;
596 yield* sql`INSERT INTO auth_devices(device_id, account_id, environment_id, client_json, created_at, expires_at, validated_at) VALUES('archive-device', ${accountHash}, ${environmentId}, '{}', ${now.epochMilliseconds}, ${now.epochMilliseconds + 3600000}, ${now.epochMilliseconds})`;
597 const bound = yield* (yield* SessionStore.SessionStore).issue({
598 deviceId: "archive-device",
599 method: "bearer-access-token",
600 scopes: [AuthArchiveReadScope, AuthArchiveWriteScope],
601 });
602 const context = yield* Effect.context<
603 Layer.Success<typeof layers> | FileSystem.FileSystem
604 >();
605 const app = yield* Effect.acquireRelease(
606 Effect.sync(() => HttpRouter.toWebHandler(ArchiveHttp.layer, { disableLogger: true })),
607 (app) => Effect.promise(() => app.dispose()),
608 );
609 const request = (
610 method: string,
611 suffix: string,
612 token?: string,
613 body?: unknown,
614 range?: string,
615 ) =>
616 Effect.promise(() =>
617 app.handler(
618 new Request(
619 `https://archive.example.test${suffix.startsWith("/api/archive/") ? suffix : `/api/archive/catalog${suffix}`}`,
620 {
621 method,
622 headers: {
623 ...(token ? { authorization: `Bearer ${token}` } : {}),
624 ...(range ? { range } : {}),
625 ...(body ? { "content-type": "application/json" } : {}),
626 },
627 ...(body ? { body: JSON.stringify(body) } : {}),
628 },
629 ),
630 context,
631 ),
632 );
633 assert.strictEqual((yield* request("GET", "")).status, 401);
634 assert.strictEqual((yield* request("GET", "", legacy.token)).status, 403);
635 assert.strictEqual((yield* request("GET", "", write.token)).status, 403);
636 const response = yield* request("GET", "", read.token);
637 assert.strictEqual(response.status, 200);
638 const catalog = yield* decodeCatalog(yield* Effect.promise(() => response.json()));
639 const entry = catalog.entries[0]!;
640 const publication = {
641 environmentId,
642 manifestHash: entry.manifestHash,
643 project: entry.project,
644 };
645 assert.strictEqual((yield* request("POST", "", read.token, publication)).status, 403);
646 assert.strictEqual((yield* request("POST", "", write.token, publication)).status, 403);
647 assert.strictEqual(
648 (yield* request("POST", "", bound.token, {
649 ...publication,
650 environmentId: "other-environment",
651 })).status,
652 403,
653 );
654 assert.strictEqual((yield* request("POST", "", bound.token, publication)).status, 200);
655 const suffix = `/${encodeURIComponent(environmentId)}/${threadId}`;
656 const detail = yield* request("GET", suffix, read.token);
657 assert.strictEqual(detail.status, 200);
658 const snapshot = yield* decodeSnapshot(yield* Effect.promise(() => detail.json()));
659 assert.strictEqual(
660 snapshot.projection.messages[0]?.text,
661 "private archived message body",
662 );
663 const assetInput = {
664 environmentId,
665 resource: { _tag: "attachment", attachmentId: imageId, fileName: "picture.png" },
666 };
667 assert.strictEqual(
668 (yield* request("POST", "/api/archive/assets", write.token, assetInput)).status,
669 403,
670 );
671 assert.strictEqual(
672 (yield* request("POST", "/api/archive/assets", read.token, {
673 ...assetInput,
674 environmentId: "other-environment",
675 })).status,
676 503,
677 );
678 assert.strictEqual(
679 (yield* request("POST", "/api/archive/assets", read.token, {
680 environmentId,
681 resource: { _tag: "media-file", threadId, path: "/etc/passwd" },
682 })).status,
683 503,
684 );
685 const issued = yield* request("POST", "/api/archive/assets", bound.token, assetInput);
686 assert.strictEqual(issued.status, 200);
687 const imageUrl = (yield* decodeAssetUrl(yield* Effect.promise(() => issued.json())))
688 .relativeUrl;
689 const image = yield* request("GET", imageUrl);
690 assert.strictEqual(image.status, 200);
691 assert.strictEqual(image.headers.get("content-type"), "image/png");
692 assert.strictEqual(image.headers.get("cache-control"), "private, no-store");
693 assert.deepEqual(
694 new Uint8Array(yield* Effect.promise(() => image.arrayBuffer())),
695 new Uint8Array([137, 80, 78, 71, 1, 2, 3]),
696 );
697 const movieIssued = yield* request("POST", "/api/archive/assets", bound.token, {
698 environmentId,
699 resource: { _tag: "attachment", attachmentId: movieId, fileName: "movie.mp4" },
700 });
701 assert.strictEqual(movieIssued.status, 200);
702 const movieUrl = (yield* decodeAssetUrl(yield* Effect.promise(() => movieIssued.json())))
703 .relativeUrl;
704 const movie = yield* request("GET", movieUrl, undefined, undefined, "bytes=3-6");
705 assert.strictEqual(movie.status, 206);
706 assert.strictEqual(movie.headers.get("content-type"), "video/mp4");
707 assert.strictEqual(movie.headers.get("content-range"), "bytes 3-6/10");
708 assert.strictEqual(yield* Effect.promise(() => movie.text()), "3456");
709 assert.strictEqual(
710 (yield* request("GET", movieUrl, undefined, undefined, "bytes=20-")).status,
711 416,
712 );
713 yield* auth.revokeSession(bound.sessionId);
714 assert.strictEqual((yield* request("GET", imageUrl)).status, 404);
715 const renewed = yield* (yield* SessionStore.SessionStore).issue({
716 deviceId: "archive-device",
717 method: "bearer-access-token",
718 scopes: [AuthArchiveReadScope, AuthArchiveWriteScope],
719 });
720 const renewedAsset = yield* request(
721 "POST",
722 "/api/archive/assets",
723 renewed.token,
724 assetInput,
725 );
726 const renewedUrl = (yield* decodeAssetUrl(
727 yield* Effect.promise(() => renewedAsset.json()),
728 )).relativeUrl;
729 yield* TestClock.adjust("5 minutes");
730 assert.strictEqual((yield* request("GET", renewedUrl)).status, 404);
731 assert.strictEqual(
732 (yield* request("DELETE", suffix, read.token, { manifestHash: entry.manifestHash }))
733 .status,
734 403,
735 );
736 assert.strictEqual(
737 (yield* request("DELETE", suffix, renewed.token, { manifestHash: "0".repeat(64) }))
738 .status,
739 204,
740 );
741 assert.strictEqual(
742 (yield* (yield* ArchiveCatalog.ArchiveCatalog).list).entries.length,
743 1,
744 );
745 assert.strictEqual(
746 (yield* request("DELETE", suffix, renewed.token, { manifestHash: entry.manifestHash }))
747 .status,
748 204,
749 );
750 assert.strictEqual(
751 (yield* (yield* ArchiveCatalog.ArchiveCatalog).list).entries.length,
752 0,
753 );
754 yield* sql`UPDATE auth_devices SET revoked_at = ${now.epochMilliseconds} WHERE device_id = 'archive-device'`;
755 assert.strictEqual((yield* request("POST", "", renewed.token, publication)).status, 401);
756 }).pipe(Effect.provide(layers));
757 }),
758 );
759
760 it.effect(
761 "rejects an older publication and keeps a newer archive when an old withdrawal arrives",
762 () =>
763 Effect.gen(function* () {
764 const f = yield* fixture();
765 yield* Effect.gen(function* () {
766 yield* seed;
767 const archive = yield* TranscriptArchive.TranscriptArchive;
768 const catalog = yield* ArchiveCatalog.ArchiveCatalog;
769 const client = yield* ArchiveClient.ArchiveClient;
770 yield* archive.offload(threadId);
771 const old = (yield* catalog.list).entries[0]!;
772 yield* archive.restore(threadId);
773 const message = (yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadProjection(
774 threadId,
775 )).messages[0]!;
776 yield* (yield* EventSink.EventSinkV2).write({
777 events: [
778 {
779 id: EventId.make("newer-archive"),
780 type: "message.updated",
781 threadId,
782 occurredAt: yield* DateTime.now,
783 payload: { ...message, text: "newer archived body" },
784 },
785 ],
786 });
787 yield* archive.offload(threadId);
788 const current = (yield* catalog.list).entries[0]!;
789 assert.isAbove(current.throughSequence, old.throughSequence);
790 assert.strictEqual(
791 (yield* client
792 .publish({
793 environmentId: old.environmentId,
794 manifestHash: old.manifestHash,
795 project: old.project,
796 })
797 .pipe(Effect.flip)).reason,
798 "changed",
799 );
800 yield* catalog.withdraw(old.environmentId, threadId, old.manifestHash);
801 assert.strictEqual((yield* catalog.list).entries[0]?.manifestHash, current.manifestHash);
802 assert.strictEqual(
803 (yield* catalog.read(current.environmentId, threadId)).projection.messages[0]?.text,
804 "newer archived body",
805 );
806 }).pipe(Effect.provide(f.layer));
807 }),
808 );
809
810 it.effect(
811 "rejects an absent or wrong provisioned storage identity before creating archive directories",
812 () =>
813 Effect.gen(function* () {
814 const fs = yield* FileSystem.FileSystem;
815 const path = yield* Path.Path;
816 const root = yield* fs.makeTempDirectoryScoped({ prefix: "knife-archive-identity-" });
817 const layer = ArchiveObjects.layer.pipe(
818 Layer.provide(
819 ConfigProvider.layer(
820 ConfigProvider.fromUnknown({
821 T3CODE_ARCHIVE_DIR: root,
822 T3CODE_ARCHIVE_STORE_ID: "test-store",
823 }),
824 ),
825 ),
826 );
827 yield* Effect.gen(function* () {
828 const objects = yield* ArchiveObjects.ArchiveObjects;
829 assert.strictEqual((yield* objects.directory.pipe(Effect.flip)).reason, "storage");
830 assert.isFalse(yield* fs.exists(path.join(root, "objects")));
831 yield* fs.writeFileString(path.join(root, ".knife-archive-store-id"), "different-store");
832 assert.strictEqual((yield* objects.directory.pipe(Effect.flip)).reason, "storage");
833 assert.isFalse(yield* fs.exists(path.join(root, "objects")));
834 yield* fs.writeFileString(path.join(root, ".knife-archive-store-id"), "test-store\n");
835 assert.strictEqual(yield* objects.directory, path.join(root, "objects"));
836 }).pipe(Effect.provide(layer));
837 }),
838 );
839
840 it.effect("retains archived payloads while a run is queued, running, or awaiting input", () =>
841 Effect.gen(function* () {
842 const f = yield* fixture();
843 yield* Effect.gen(function* () {
844 yield* seed;
845 const sql = yield* SqlClient.SqlClient;
846 const archive = yield* TranscriptArchive.TranscriptArchive;
847 for (const status of ["preparing", "queued", "starting", "running", "waiting"]) {
848 yield* sql`INSERT INTO orchestration_v2_projection_runs(run_id, thread_id, ordinal, provider, status, requested_at, payload_json) VALUES('busy-run', ${threadId}, 1, 'codex', ${status}, '2026-01-01T00:00:00Z', '{}')`;
849 yield* archive.offload(threadId);
850 yield* archive.sweep;
851 assert.strictEqual((yield* sql`SELECT 1 FROM transcript_archives`).length, 0);
852 yield* sql`DELETE FROM orchestration_v2_projection_runs WHERE run_id = 'busy-run'`;
853 }
854 assert.strictEqual(
855 (yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadProjection(threadId))
856 .messages[0]?.text,
857 "private archived message body",
858 );
859 yield* archive.sweep;
860 assert.strictEqual((yield* sql`SELECT 1 FROM transcript_archives`).length, 1);
861 }).pipe(Effect.provide(f.layer));
862 }),
863 );
864
865 it.effect("rejects a corrupt archive without unarchiving or replacing local indexes", () =>
866 Effect.gen(function* () {
867 const f = yield* fixture();
868 yield* Effect.gen(function* () {
869 const thread = yield* seed;
870 const archive = yield* TranscriptArchive.TranscriptArchive;
871 const sql = yield* SqlClient.SqlClient;
872 const fs = yield* FileSystem.FileSystem;
873 const projections = yield* ProjectionStore.ProjectionStoreV2;
874 yield* archive.offload(threadId);
875 const entry = (yield* sql<{
876 object_hash: string;
877 }>`SELECT object_hash FROM transcript_archives WHERE thread_id = ${threadId}`)[0]!;
878 yield* fs.writeFileString(f.path.join(f.root, "objects", entry.object_hash), "corrupt");
879 assert.isTrue(
880 (yield* projections.getThreadProjection(threadId).pipe(Effect.exit))._tag === "Failure",
881 );
882 const result = yield* (yield* EventSink.EventSinkV2)
883 .write({
884 events: [
885 {
886 id: EventId.make("corrupt-unarchive"),
887 type: "thread.unarchived",
888 threadId,
889 occurredAt: yield* DateTime.now,
890 payload: thread,
891 },
892 ],
893 })
894 .pipe(Effect.exit);
895 assert.strictEqual(result._tag, "Failure");
896 assert.isNotNull((yield* projections.getThreadShell(threadId))?.archivedAt);
897 assert.strictEqual(
898 (yield* sql`SELECT 1 FROM transcript_archives WHERE thread_id = ${threadId}`).length,
899 1,
900 );
901 const message = (yield* sql<{
902 payload_json: string;
903 }>`SELECT payload_json FROM orchestration_v2_projection_messages WHERE thread_id = ${threadId}`)[0]!;
904 assert.isFalse(message.payload_json.includes("private archived"));
905 }).pipe(Effect.provide(f.layer));
906 }),
907 );
908
909 it.effect("does not acknowledge an object when its durable file flush fails", () =>
910 Effect.gen(function* () {
911 const f = yield* fixture();
912 yield* Effect.gen(function* () {
913 const fs = yield* FileSystem.FileSystem;
914 const failure = PlatformError.systemError({
915 _tag: "Unknown",
916 module: "FileSystem",
917 method: "sync",
918 });
919 const objects = yield* ArchiveObjects.ArchiveObjects.pipe(
920 Effect.provide(
921 Layer.fresh(ArchiveObjects.layer).pipe(
922 Layer.provide(
923 ConfigProvider.layer(ConfigProvider.fromUnknown({ T3CODE_ARCHIVE_DIR: f.root })),
924 ),
925 ),
926 ),
927 Effect.provideService(FileSystem.FileSystem, {
928 ...fs,
929 open: (file, options) =>
930 fs
931 .open(file, options)
932 .pipe(Effect.map((handle) => ({ ...handle, sync: Effect.fail(failure) }))),
933 }),
934 );
935 const bytes = new TextEncoder().encode("not durable");
936 const hash = Buffer.from(yield* (yield* Crypto.Crypto).digest("SHA-256", bytes)).toString(
937 "hex",
938 );
939 assert.strictEqual(
940 (yield* objects.put(hash, Stream.succeed(bytes)).pipe(Effect.flip)).reason,
941 "storage",
942 );
943 assert.isFalse(yield* fs.exists(f.path.join(f.root, "objects", hash)));
944 }).pipe(Effect.provide(f.layer));
945 }),
946 );
947
948 it.effect("retains source payloads when the archive store identity is absent or mismatched", () =>
949 Effect.gen(function* () {
950 for (const mode of ["unconfigured", "missing", "mismatch"] as const) {
951 const f = yield* fixture();
952 yield* Effect.gen(function* () {
953 yield* seed;
954 const fs = yield* FileSystem.FileSystem;
955 const client = yield* ArchiveClient.ArchiveClient;
956 let archive = yield* TranscriptArchive.TranscriptArchive;
957 if (mode === "unconfigured") {
958 const catalog = yield* ArchiveCatalog.ArchiveCatalog.pipe(
959 Effect.provide(
960 Layer.fresh(
961 ArchiveCatalog.layer.pipe(
962 Layer.provide(ArchiveObjects.layer),
963 Layer.provide(
964 ConfigProvider.layer(
965 ConfigProvider.fromUnknown({ T3CODE_ARCHIVE_DIR: f.root }),
966 ),
967 ),
968 ),
969 ),
970 ),
971 );
972 archive = yield* TranscriptArchive.TranscriptArchive.pipe(
973 Effect.provide(Layer.fresh(TranscriptArchive.layer)),
974 Effect.provideService(ArchiveClient.ArchiveClient, {
975 ...client,
976 publish: catalog.publish,
977 }),
978 );
979 } else if (mode === "missing")
980 yield* fs.remove(f.path.join(f.root, ".knife-archive-store-id"));
981 else
982 yield* fs.writeFileString(
983 f.path.join(f.root, ".knife-archive-store-id"),
984 "wrong-store",
985 );
986 assert.strictEqual(
987 (yield* archive.offload(threadId).pipe(Effect.flip)).reason,
988 "storage",
989 );
990 assert.strictEqual(
991 (yield* (yield* SqlClient.SqlClient)`SELECT 1 FROM transcript_archives`).length,
992 0,
993 );
994 assert.strictEqual(
995 (yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadProjection(threadId))
996 .messages[0]?.text,
997 "private archived message body",
998 );
999 }).pipe(Effect.provide(f.layer));
1000 }
1001 }),
1002 );
1003
1004 it.effect(
1005 "keeps existing attachment URLs playable after eviction and restores bytes on unarchive",
1006 () =>
1007 Effect.gen(function* () {
1008 const f = yield* fixture();
1009 yield* Effect.gen(function* () {
1010 const thread = yield* seed;
1011 const fs = yield* FileSystem.FileSystem;
1012 const config = yield* ServerConfig.ServerConfig;
1013 const archive = yield* TranscriptArchive.TranscriptArchive;
1014 const projections = yield* ProjectionStore.ProjectionStoreV2;
1015 const sink = yield* EventSink.EventSinkV2;
1016 const now = yield* DateTime.now;
1017 const id = ChatAttachmentId.make(createDeterministicAttachmentId(threadId, "recording")!);
1018 const file = f.path.join(config.attachmentsDir, `${id}.png`);
1019 yield* fs.makeDirectory(config.attachmentsDir, { recursive: true });
1020 const bytes = new Uint8Array([137, 80, 78, 71, 1, 2, 3]);
1021 yield* fs.writeFile(file, bytes);
1022 const message = (yield* projections.getThreadProjection(threadId)).messages[0]!;
1023 yield* sink.write({
1024 events: [
1025 {
1026 id: EventId.make("attachment-added"),
1027 type: "message.updated",
1028 threadId,
1029 occurredAt: now,
1030 payload: {
1031 ...message,
1032 attachments: [
1033 {
1034 type: "image",
1035 id,
1036 name: "recording.png",
1037 mimeType: "image/png",
1038 sizeBytes: bytes.length,
1039 },
1040 ],
1041 },
1042 },
1043 ],
1044 });
1045 const issue = issueAssetUrl({
1046 resource: {
1047 _tag: "attachment",
1048 attachmentId: id,
1049 fileName: "recording.png",
1050 mimeType: "image/png",
1051 },
1052 });
1053 const old = yield* issue;
1054 const resolve = (url: string) => {
1055 const suffix = url.slice(`${ASSET_ROUTE_PREFIX}/`.length);
1056 const separator = suffix.indexOf("/");
1057 return resolveAsset(suffix.slice(0, separator), suffix.slice(separator + 1));
1058 };
1059 assert.strictEqual((yield* resolve(old.relativeUrl))?.kind, "file");
1060 yield* archive.sweep;
1061 assert.isFalse(yield* fs.exists(file));
1062 for (const url of [old.relativeUrl, (yield* issue).relativeUrl]) {
1063 const asset = yield* resolve(url);
1064 assert.strictEqual(asset?.kind, "archived-file");
1065 if (asset?.kind !== "archived-file")
1066 return yield* Effect.die("Expected archived asset");
1067 const opened = yield* (yield* ArchiveClient.ArchiveClient).open(
1068 asset.location,
1069 asset.hash,
1070 );
1071 assert.strictEqual(opened.kind, "file");
1072 if (opened.kind === "file")
1073 assert.deepStrictEqual(yield* fs.readFile(opened.path), bytes);
1074 }
1075 yield* sink.write({
1076 events: [
1077 {
1078 id: EventId.make("attachment-unarchive"),
1079 type: "thread.unarchived",
1080 threadId,
1081 occurredAt: now,
1082 payload: thread,
1083 },
1084 ],
1085 });
1086 assert.deepStrictEqual(yield* fs.readFile(file), bytes);
1087 assert.strictEqual((yield* resolve(old.relativeUrl))?.kind, "file");
1088 }).pipe(
1089 Effect.provide(
1090 Layer.mergeAll(
1091 f.layer,
1092 WorkspacePaths.layer,
1093 Layer.mock(NativeAppIconResolver.NativeAppIconResolver)({}),
1094 Layer.mock(ProjectFaviconResolver.ProjectFaviconResolver)({}),
1095 Layer.mock(Orchestrator.OrchestratorV2)({}),
1096 ),
1097 ),
1098 );
1099 }),
1100 );
1101
1102 it.effect(
1103 "keeps fork history readable and its source attachment warm through offline rebuild",
1104 () =>
1105 Effect.gen(function* () {
1106 const f = yield* fixture();
1107 yield* Effect.gen(function* () {
1108 const thread = yield* seed;
1109 const projections = yield* ProjectionStore.ProjectionStoreV2;
1110 const sink = yield* EventSink.EventSinkV2;
1111 const archive = yield* TranscriptArchive.TranscriptArchive;
1112 const fs = yield* FileSystem.FileSystem;
1113 const sql = yield* SqlClient.SqlClient;
1114 const now = yield* DateTime.now;
1115 const forkId = ThreadId.make("archive-fork");
1116 const runId = RunId.make("archive-source-run");
1117 const nodeId = NodeId.make("archive-source-node");
1118 const message = (yield* projections.getThreadProjection(threadId)).messages[0]!;
1119 const attachmentId = ChatAttachmentId.make(
1120 createDeterministicAttachmentId(threadId, "source")!,
1121 );
1122 const config = yield* ServerConfig.ServerConfig;
1123 const attachment = f.path.join(config.attachmentsDir, `${attachmentId}.png`);
1124 yield* fs.makeDirectory(config.attachmentsDir, { recursive: true });
1125 yield* fs.writeFileString(attachment, "source attachment");
1126 const fork = {
1127 ...thread,
1128 id: forkId,
1129 forkedFrom: { type: "run" as const, threadId, runId },
1130 lineage: {
1131 parentThreadId: threadId,
1132 relationshipToParent: "fork" as const,
1133 rootThreadId: threadId,
1134 },
1135 };
1136 const item = (yield* projections.getThreadProjection(threadId)).turnItems[0]!;
1137 yield* sink.write({
1138 events: [
1139 {
1140 id: EventId.make("source-run"),
1141 type: "run.created",
1142 threadId,
1143 runId,
1144 nodeId,
1145 driver: ProviderDriverKind.make("codex"),
1146 occurredAt: now,
1147 payload: {
1148 id: runId,
1149 threadId,
1150 ordinal: 1,
1151 providerInstanceId: thread.providerInstanceId,
1152 modelSelection: thread.modelSelection,
1153 providerThreadId: null,
1154 userMessageId: message.id,
1155 rootNodeId: nodeId,
1156 activeAttemptId: null,
1157 status: "completed",
1158 requestedAt: now,
1159 startedAt: now,
1160 completedAt: now,
1161 checkpointId: null,
1162 contextHandoffId: null,
1163 },
1164 },
1165 {
1166 id: EventId.make("source-run-message"),
1167 type: "message.updated",
1168 threadId,
1169 occurredAt: now,
1170 payload: {
1171 ...message,
1172 runId,
1173 attachments: [
1174 {
1175 type: "image",
1176 id: attachmentId,
1177 name: "source.png",
1178 mimeType: "image/png",
1179 sizeBytes: 17,
1180 },
1181 ],
1182 },
1183 },
1184 {
1185 id: EventId.make("source-run-item"),
1186 type: "turn-item.updated",
1187 threadId,
1188 occurredAt: now,
1189 payload: { ...item, runId },
1190 },
1191 {
1192 id: EventId.make("fork-created"),
1193 type: "thread.created",
1194 threadId: forkId,
1195 occurredAt: now,
1196 payload: fork,
1197 },
1198 ],
1199 });
1200 const before = yield* projections.getThreadProjection(forkId);
1201 assert.isTrue(
1202 before.visibleTurnItems.some(
1203 (item) =>
1204 item.item.type === "assistant_message" &&
1205 item.item.text === "private archived message body",
1206 ),
1207 );
1208 yield* archive.sweep;
1209 assert.isTrue(yield* fs.exists(attachment));
1210 assert.deepStrictEqual(yield* projections.getThreadProjection(forkId), before);
1211 const objects = f.path.join(f.root, "objects");
1212 yield* fs.rename(objects, `${objects}-offline`);
1213 assert.isTrue(
1214 (yield* (yield* ProjectionMaintenance.ProjectionMaintenanceV2).rebuild).valid,
1215 );
1216 yield* fs.rename(`${objects}-offline`, objects);
1217 assert.deepStrictEqual(yield* projections.getThreadProjection(forkId), before);
1218 yield* sink.write({
1219 events: [
1220 {
1221 id: EventId.make("fork-archived"),
1222 type: "thread.archived",
1223 threadId: forkId,
1224 occurredAt: now,
1225 payload: { ...fork, archivedAt: now },
1226 },
1227 ],
1228 });
1229 yield* archive.sweep;
1230 assert.isFalse(yield* fs.exists(attachment));
1231 const sourceManifest = (yield* sql<{
1232 object_hash: string;
1233 }>`SELECT object_hash FROM transcript_archives WHERE thread_id = ${threadId}`)[0]!;
1234 const forkManifest = (yield* sql<{
1235 object_hash: string;
1236 }>`SELECT object_hash FROM transcript_archives WHERE thread_id = ${forkId}`)[0]!;
1237 const source = JSON.parse(
1238 yield* fs.readFileString(f.path.join(objects, sourceManifest.object_hash)),
1239 );
1240 const savedFork = JSON.parse(
1241 yield* fs.readFileString(f.path.join(objects, forkManifest.object_hash)),
1242 );
1243 assert.strictEqual(
1244 savedFork.context.find(
1245 (s: { table: string }) => s.table === "orchestration_v2_projection_messages",
1246 ).hash,
1247 source.segments.find((s: { table: string }) => s.table === "messages").hash,
1248 );
1249 assert.deepStrictEqual(savedFork.assets, source.assets);
1250 assert.deepStrictEqual(
1251 (yield* projections.getThreadProjection(forkId)).visibleTurnItems,
1252 before.visibleTurnItems,
1253 );
1254 }).pipe(Effect.provide(f.layer));
1255 }),
1256 );
1257
1258 it.effect(
1259 "offloads bodies only after verification, reads without retaining them, rebuilds offline, and restores before unarchive",
1260 () =>
1261 Effect.gen(function* () {
1262 const f = yield* fixture();
1263 yield* Effect.gen(function* () {
1264 const thread = yield* seed;
1265 const archive = yield* TranscriptArchive.TranscriptArchive;
1266 const projections = yield* ProjectionStore.ProjectionStoreV2;
1267 const sql = yield* SqlClient.SqlClient;
1268 const events = yield* ApplicationEvents.OrchestrationEventStore;
1269 const before = yield* projections.getThreadProjection(threadId);
1270 const highWater = yield* events.latestAgentSequence(threadId);
1271 yield* archive.offload(threadId);
1272 const raw = yield* sql<{
1273 payload_json: string;
1274 }>`SELECT payload_json FROM orchestration_v2_projection_messages WHERE thread_id = ${threadId}`;
1275 assert.isFalse(raw[0]?.payload_json.includes("private archived"));
1276 const shell = yield* projections.getThreadShell(threadId);
1277 assert.strictEqual(shell?.title, "Searchable archive title");
1278 assert.strictEqual(shell?.visibleItemCount, before.visibleTurnItems.length);
1279 assert.strictEqual(yield* events.latestAgentSequence(threadId), highWater);
1280 assert.isFalse(
1281 (yield* events.getAgentReplayStats({
1282 threadId,
1283 afterSequence: 0,
1284 throughSequence: highWater,
1285 maxEvents: 100,
1286 })).hasCreateEvent,
1287 );
1288 assert.deepStrictEqual(yield* projections.getThreadProjection(threadId), before);
1289 assert.strictEqual(
1290 (yield* sql`SELECT 1 FROM transcript_archives WHERE thread_id = ${threadId}`).length,
1291 1,
1292 );
1293 assert.isFalse(
1294 (yield* sql<{
1295 payload_json: string;
1296 }>`SELECT payload_json FROM orchestration_v2_projection_messages WHERE thread_id = ${threadId}`)[0]?.payload_json.includes(
1297 "private archived",
1298 ),
1299 );
1300 const fs = yield* FileSystem.FileSystem;
1301 const objects = f.path.join(f.root, "objects");
1302 yield* fs.rename(objects, `${objects}-offline`);
1303 assert.isTrue(
1304 (yield* (yield* ProjectionMaintenance.ProjectionMaintenanceV2).rebuild).valid,
1305 );
1306 assert.isTrue(
1307 (yield* projections.getThreadProjection(threadId).pipe(Effect.exit))._tag === "Failure",
1308 );
1309 assert.strictEqual(
1310 (yield* projections.getThreadShell(threadId))?.title,
1311 "Searchable archive title",
1312 );
1313 yield* fs.rename(`${objects}-offline`, objects);
1314 const now = yield* DateTime.now;
1315 yield* (yield* EventSink.EventSinkV2).write({
1316 events: [
1317 {
1318 id: EventId.make("archive-unarchived"),
1319 type: "thread.unarchived",
1320 threadId,
1321 occurredAt: now,
1322 payload: thread,
1323 },
1324 ],
1325 });
1326 assert.strictEqual(
1327 (yield* sql`SELECT 1 FROM transcript_archives WHERE thread_id = ${threadId}`).length,
1328 0,
1329 );
1330 assert.strictEqual(
1331 (yield* projections.getThreadProjection(threadId)).messages[0]?.text,
1332 "private archived message body",
1333 );
1334 assert.isTrue(
1335 (yield* (yield* ProjectionMaintenance.ProjectionMaintenanceV2).rebuild).valid,
1336 );
1337 assert.strictEqual(
1338 (yield* projections.getThreadProjection(threadId)).messages[0]?.text,
1339 "private archived message body",
1340 );
1341 }).pipe(Effect.provide(f.layer));
1342 }),
1343 );
1344
1345 it.effect("never evicts local bodies when the destination is missing", () =>
1346 Effect.gen(function* () {
1347 const f = yield* fixture();
1348 yield* Effect.gen(function* () {
1349 yield* seed;
1350 const fs = yield* FileSystem.FileSystem;
1351 yield* fs.remove(f.root, { recursive: true });
1352 assert.strictEqual(
1353 (yield* (yield* TranscriptArchive.TranscriptArchive).offload(threadId).pipe(Effect.flip))
1354 .reason,
1355 "storage",
1356 );
1357 assert.strictEqual(
1358 (yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadProjection(threadId))
1359 .messages[0]?.text,
1360 "private archived message body",
1361 );
1362 assert.strictEqual(
1363 (yield* (yield* SqlClient.SqlClient)`SELECT 1 FROM transcript_archives`).length,
1364 0,
1365 );
1366 }).pipe(Effect.provide(f.layer));
1367 }),
1368 );
1369
1370 it.effect("rejects a wrong object hash without publishing partial bytes", () =>
1371 Effect.gen(function* () {
1372 const f = yield* fixture();
1373 yield* Effect.gen(function* () {
1374 const store = yield* ArchiveObjects.ArchiveObjects;
1375 const hash = "0".repeat(64);
1376 const error = yield* store
1377 .put(hash, Stream.make(new TextEncoder().encode("wrong bytes")))
1378 .pipe(Effect.flip);
1379 assert.strictEqual(error.reason, "integrity");
1380 assert.isFalse(
1381 yield* (yield* FileSystem.FileSystem).exists(f.path.join(f.root, "objects", hash)),
1382 );
1383 }).pipe(Effect.provide(f.layer));
1384 }),
1385 );
1386 it.effect("an unarchive committed during upload wins over offloading", () =>
1387 Effect.gen(function* () {
1388 const f = yield* fixture();
1389 yield* Effect.gen(function* () {
1390 const thread = yield* seed;
1391 const uploaded = yield* Deferred.make<void>();
1392 const resume = yield* Deferred.make<void>();
1393 const client = yield* ArchiveClient.ArchiveClient;
1394 const archive = yield* TranscriptArchive.TranscriptArchive.pipe(
1395 Effect.provide(Layer.fresh(TranscriptArchive.layer)),
1396 Effect.provideService(ArchiveClient.ArchiveClient, {
1397 ...client,
1398 upload: (file) =>
1399 client.upload(file).pipe(
1400 Effect.tap(() => Deferred.succeed(uploaded, undefined)),
1401 Effect.tap(() => Deferred.await(resume)),
1402 ),
1403 }),
1404 );
1405 const offload = yield* archive.offload(threadId).pipe(Effect.flip, Effect.forkScoped);
1406 yield* Effect.raceFirst(
1407 Deferred.await(uploaded),
1408 Fiber.await(offload).pipe(Effect.flatMap((exit) => Effect.die(exit))),
1409 );
1410 const now = yield* DateTime.now;
1411 yield* (yield* EventSink.EventSinkV2).write({
1412 events: [
1413 {
1414 id: EventId.make("race-unarchive"),
1415 type: "thread.unarchived",
1416 threadId,
1417 occurredAt: now,
1418 payload: thread,
1419 },
1420 ],
1421 });
1422 yield* Deferred.succeed(resume, undefined);
1423 assert.strictEqual((yield* Fiber.join(offload)).reason, "changed");
1424 assert.strictEqual(
1425 (yield* (yield* SqlClient.SqlClient)`SELECT 1 FROM transcript_archives`).length,
1426 0,
1427 );
1428 assert.strictEqual(
1429 (yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadProjection(threadId))
1430 .messages[0]?.text,
1431 "private archived message body",
1432 );
1433 yield* archive.sweep;
1434 assert.strictEqual((yield* (yield* ArchiveCatalog.ArchiveCatalog).list).entries.length, 0);
1435 }).pipe(Effect.provide(f.layer));
1436 }),
1437 );
1438});