From bb40e9a3f97e2bd3c9d1193ef9f85a3a908bd4e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Tue, 14 Jul 2026 12:38:48 +0100 Subject: [PATCH 01/10] Add cold storage for archived conversations --- .../archive/ArchivedThreadsRouteScreen.tsx | 24 +- .../archive/ArchivedThreadsScreen.tsx | 20 +- .../src/features/home/useThreadListActions.ts | 6 +- apps/server/src/config.ts | 3 + .../Layers/OrchestrationEngine.ts | 30 ++ .../Layers/ThreadColdStorage.test.ts | 131 +++++ .../orchestration/Layers/ThreadColdStorage.ts | 491 ++++++++++++++++++ .../Layers/ThreadDeletionReactor.ts | 82 ++- .../Services/ThreadColdStorage.ts | 38 ++ .../Services/ThreadDeletionReactor.ts | 4 +- apps/server/src/persistence/Migrations.ts | 4 + .../033_034_ThreadStorageLifecycle.test.ts | 69 +++ .../Migrations/033_ThreadColdArchive.ts | 68 +++ .../034_DeletedThreadCleanupQueue.ts | 24 + .../src/provider/Layers/EventNdjsonLogger.ts | 19 + .../src/provider/acp/AcpNativeLogging.test.ts | 1 + apps/server/src/server.ts | 7 +- apps/web/src/components/Sidebar.tsx | 15 +- .../components/settings/SettingsPanels.tsx | 93 ++-- apps/web/src/hooks/useThreadActions.ts | 23 +- apps/web/src/optimisticThreadArchiveStore.ts | 33 ++ 21 files changed, 1114 insertions(+), 71 deletions(-) create mode 100644 apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts create mode 100644 apps/server/src/orchestration/Layers/ThreadColdStorage.ts create mode 100644 apps/server/src/orchestration/Services/ThreadColdStorage.ts create mode 100644 apps/server/src/persistence/Migrations/033_034_ThreadStorageLifecycle.test.ts create mode 100644 apps/server/src/persistence/Migrations/033_ThreadColdArchive.ts create mode 100644 apps/server/src/persistence/Migrations/034_DeletedThreadCleanupQueue.ts create mode 100644 apps/web/src/optimisticThreadArchiveStore.ts diff --git a/apps/mobile/src/features/archive/ArchivedThreadsRouteScreen.tsx b/apps/mobile/src/features/archive/ArchivedThreadsRouteScreen.tsx index c2381ef2580..fc20c581ad9 100644 --- a/apps/mobile/src/features/archive/ArchivedThreadsRouteScreen.tsx +++ b/apps/mobile/src/features/archive/ArchivedThreadsRouteScreen.tsx @@ -16,6 +16,8 @@ import { refreshArchivedThreadsForEnvironment, useArchivedThreadSnapshots, } from "./useArchivedThreadSnapshots"; +import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; +import { scopedThreadKey } from "../../lib/scopedEntities"; export function ArchivedThreadsRouteScreen() { const { expand } = useClerkSettingsSheetDetent(); @@ -23,6 +25,9 @@ export function ArchivedThreadsRouteScreen() { const [searchQuery, setSearchQuery] = useState(""); const [selectedEnvironmentId, setSelectedEnvironmentId] = useState(null); const [sortOrder, setSortOrder] = useState("newest"); + const [unarchivingThreadKeys, setUnarchivingThreadKeys] = useState>( + () => new Set(), + ); const environments = useMemo>( () => Arr.sort( @@ -67,6 +72,22 @@ export function ArchivedThreadsRouteScreen() { ); const { unarchiveThread, confirmDeleteThread } = useArchivedThreadListActions(refreshChangedEnvironment); + const handleUnarchiveThread = useCallback( + async (thread: EnvironmentThreadShell) => { + const threadKey = scopedThreadKey(thread.environmentId, thread.id); + setUnarchivingThreadKeys((current) => new Set(current).add(threadKey)); + try { + await unarchiveThread(thread); + } finally { + setUnarchivingThreadKeys((current) => { + const next = new Set(current); + next.delete(threadKey); + return next; + }); + } + }, + [unarchiveThread], + ); useFocusEffect( useCallback(() => { @@ -86,10 +107,11 @@ export function ArchivedThreadsRouteScreen() { onRefresh={refresh} onSearchQueryChange={setSearchQuery} onSortOrderChange={setSortOrder} - onUnarchiveThread={unarchiveThread} + onUnarchiveThread={(thread) => void handleUnarchiveThread(thread)} searchQuery={searchQuery} selectedEnvironmentId={selectedEnvironmentId} sortOrder={sortOrder} + unarchivingThreadKeys={unarchivingThreadKeys} /> ); } diff --git a/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx b/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx index 916802e9faf..a36d39b73ea 100644 --- a/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx +++ b/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx @@ -31,6 +31,7 @@ import { useThemeColor } from "../../lib/useThemeColor"; import { ThreadSwipeable } from "../home/thread-swipe-actions"; import { createNativeMailSearchToolbarItem } from "../layout/native-mail-search-toolbar"; import type { ArchivedThreadGroup, ArchivedThreadSortOrder } from "./archivedThreadList"; +import { scopedThreadKey } from "../../lib/scopedEntities"; export interface ArchivedThreadsHeaderEnvironment { readonly environmentId: EnvironmentId; @@ -383,6 +384,7 @@ function ArchivedThreadRow(props: { readonly environmentLabel: string | null; readonly isFirst: boolean; readonly isLast: boolean; + readonly isUnarchiving: boolean; readonly onDelete: () => void; readonly onSwipeableClose: (methods: SwipeableMethods) => void; readonly onSwipeableWillOpen: (methods: SwipeableMethods) => void; @@ -420,7 +422,7 @@ function ArchivedThreadRow(props: { accessibilityLabel: `Unarchive ${props.thread.title}`, icon: "arrow.uturn.backward", label: "Unarchive", - onPress: props.onUnarchive, + onPress: props.isUnarchiving ? () => undefined : props.onUnarchive, }} simultaneousWithExternalGesture={props.simultaneousSwipeGesture} threadTitle={props.thread.title} @@ -434,7 +436,16 @@ function ArchivedThreadRow(props: { }} > - + {props.isUnarchiving ? ( + + ) : ( + + )} @@ -500,6 +511,7 @@ export function ArchivedThreadsScreen(props: { readonly onSearchQueryChange: (query: string) => void; readonly onSortOrderChange: (sortOrder: ArchivedThreadSortOrder) => void; readonly onUnarchiveThread: (thread: EnvironmentThreadShell) => void; + readonly unarchivingThreadKeys: ReadonlySet; }) { const { onDeleteThread, onUnarchiveThread } = props; const openSwipeableRef = useRef(null); @@ -564,6 +576,9 @@ export function ArchivedThreadsScreen(props: { environmentLabel={item.environmentLabel} isFirst={item.isFirst} isLast={item.isLast} + isUnarchiving={props.unarchivingThreadKeys.has( + scopedThreadKey(item.thread.environmentId, item.thread.id), + )} onDelete={() => onDeleteThread(item.thread)} onSwipeableClose={handleSwipeableClose} onSwipeableWillOpen={handleSwipeableWillOpen} @@ -579,6 +594,7 @@ export function ArchivedThreadsScreen(props: { handleSwipeableWillOpen, onDeleteThread, onUnarchiveThread, + props.unarchivingThreadKeys, ], ); const listEmptyComponent = useMemo(() => { diff --git a/apps/mobile/src/features/home/useThreadListActions.ts b/apps/mobile/src/features/home/useThreadListActions.ts index 8effdc942d9..858c473c3f0 100644 --- a/apps/mobile/src/features/home/useThreadListActions.ts +++ b/apps/mobile/src/features/home/useThreadListActions.ts @@ -129,7 +129,7 @@ export function useThreadListActions(): { export function useArchivedThreadListActions( onCompleted: (thread: EnvironmentThreadShell) => void, ): { - readonly unarchiveThread: (thread: EnvironmentThreadShell) => void; + readonly unarchiveThread: (thread: EnvironmentThreadShell) => Promise; readonly confirmDeleteThread: (thread: EnvironmentThreadShell) => void; } { const handleCompleted = useCallback( @@ -140,9 +140,7 @@ export function useArchivedThreadListActions( ); const executeAction = useThreadActionExecutor(handleCompleted); const unarchiveThread = useCallback( - (thread: EnvironmentThreadShell) => { - void executeAction("unarchive", thread); - }, + (thread: EnvironmentThreadShell) => executeAction("unarchive", thread), [executeAction], ); const confirmDeleteThread = useConfirmDeleteThread(executeAction); diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index 2608ccc16ae..9bc8518b609 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -28,6 +28,7 @@ export type StartupPresentation = typeof StartupPresentation.Type; export interface ServerDerivedPaths { readonly stateDir: string; readonly dbPath: string; + readonly archiveDbPath: string; readonly keybindingsConfigPath: string; readonly settingsPath: string; readonly providerStatusCacheDir: string; @@ -95,6 +96,7 @@ export const deriveServerPaths = Effect.fn(function* ( const { join } = yield* Path.Path; const stateDir = join(baseDir, devUrl !== undefined ? "dev" : "userdata"); const dbPath = join(stateDir, "state.sqlite"); + const archiveDbPath = join(stateDir, "archive.sqlite"); const attachmentsDir = join(stateDir, "attachments"); const logsDir = join(stateDir, "logs"); const providerLogsDir = join(logsDir, "provider"); @@ -102,6 +104,7 @@ export const deriveServerPaths = Effect.fn(function* ( return { stateDir, dbPath, + archiveDbPath, keybindingsConfigPath: join(stateDir, "keybindings.json"), settingsPath: join(stateDir, "settings.json"), providerStatusCacheDir, diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 9598d1c3ef5..6fab829441e 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -45,6 +45,7 @@ import { OrchestrationEngineService, type OrchestrationEngineShape, } from "../Services/OrchestrationEngine.ts"; +import { ThreadColdStorage } from "../Services/ThreadColdStorage.ts"; const isOrchestrationCommandPreviouslyRejectedError = Schema.is( OrchestrationCommandPreviouslyRejectedError, ); @@ -82,6 +83,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { const commandReceiptRepository = yield* OrchestrationCommandReceiptRepository; const projectionPipeline = yield* OrchestrationProjectionPipeline; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + const threadColdStorage = yield* Effect.serviceOption(ThreadColdStorage); const crypto = yield* Crypto.Crypto; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); @@ -150,6 +152,24 @@ const makeOrchestrationEngine = Effect.gen(function* () { }); } + const unarchiveThreadId = + envelope.command.type === "thread.unarchive" ? envelope.command.threadId : null; + if (unarchiveThreadId !== null && Option.isSome(threadColdStorage)) { + const restored = yield* threadColdStorage.value.restoreTree(unarchiveThreadId).pipe( + Effect.mapError( + (cause) => + new OrchestrationCommandInvariantError({ + commandType: envelope.command.type, + detail: "Failed to restore the archived conversation.", + cause, + }), + ), + ); + if (restored) { + commandReadModel = yield* projectionSnapshotQuery.getCommandReadModel(); + } + } + const eventBase = yield* decideOrchestrationCommand({ command: envelope.command, readModel: commandReadModel, @@ -228,6 +248,16 @@ const makeOrchestrationEngine = Effect.gen(function* () { ); } } + if (unarchiveThreadId !== null && Option.isSome(threadColdStorage)) { + yield* threadColdStorage.value.finishRestoreTree(unarchiveThreadId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("failed to finalize restored archive bundle", { + threadId: unarchiveThreadId, + cause: Cause.pretty(cause), + }), + ), + ); + } return { sequence: committedCommand.lastSequence }; }).pipe(Effect.withSpan(`orchestration.command.${envelope.command.type}`)), ).pipe( diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts new file mode 100644 index 00000000000..15f16175c63 --- /dev/null +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts @@ -0,0 +1,131 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as ServerConfig from "../../config.ts"; +import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { ThreadColdStorage } from "../Services/ThreadColdStorage.ts"; +import { ThreadColdStorageLive } from "./ThreadColdStorage.ts"; + +const layer = it.layer( + ThreadColdStorageLive.pipe( + Layer.provideMerge(SqlitePersistenceMemory), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-cold-storage-" })), + Layer.provideMerge(NodeServices.layer), + ), +); + +layer("ThreadColdStorage", (it) => { + it.effect("compresses conversation data, destroys logs, restores content, and hard-deletes", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const threadId = ThreadId.make("thread-cold"); + const attachmentName = "thread-cold-00000000-0000-4000-8000-000000000001.png"; + + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, + interaction_mode, created_at, updated_at, archived_at + ) VALUES ( + ${threadId}, 'project-1', 'Cold thread', + '{"instanceId":"codex","model":"gpt-5.5","options":[]}', + 'full-access', 'default', '2026-07-01T00:00:00.000Z', + '2026-07-02T00:00:00.000Z', '2026-07-02T00:00:00.000Z' + ) + `; + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES ( + 'message-1', ${threadId}, NULL, 'user', 'keep this conversation', + '[{"type":"image","id":"thread-cold-00000000-0000-4000-8000-000000000001","name":"image.png","mimeType":"image/png"}]', + 0, '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z' + ) + `; + yield* sql` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, command_id, causation_event_id, correlation_id, + actor_kind, payload_json, metadata_json + ) VALUES ( + 'event-1', 'thread', ${threadId}, 1, 'thread.created', + '2026-07-01T00:00:00.000Z', 'command-1', NULL, 'command-1', + 'system', '{}', '{}' + ) + `; + + const attachmentPath = path.join(config.attachmentsDir, attachmentName); + const providerLogPath = path.join(config.providerLogsDir, "thread-cold.log"); + const rotatedProviderLogPath = `${providerLogPath}.1`; + yield* fs.writeFileString(attachmentPath, "image bytes"); + yield* fs.writeFileString(providerLogPath, "diagnostic"); + yield* fs.writeFileString(rotatedProviderLogPath, "old diagnostic"); + + yield* storage.archiveThread(threadId); + + const archivedMessageCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + const archivedEventCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM orchestration_events WHERE stream_id = ${threadId} + `; + const shellCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM projection_threads WHERE thread_id = ${threadId} + `; + const manifest = yield* sql<{ readonly status: string; readonly compressedBytes: number }>` + SELECT status, compressed_bytes AS "compressedBytes" + FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.strictEqual(archivedMessageCount[0]?.count, 0); + assert.strictEqual(archivedEventCount[0]?.count, 0); + assert.strictEqual(shellCount[0]?.count, 1); + assert.strictEqual(manifest[0]?.status, "cold"); + assert.isAbove(manifest[0]?.compressedBytes ?? 0, 0); + assert.isFalse(yield* fs.exists(attachmentPath)); + assert.isFalse(yield* fs.exists(providerLogPath)); + assert.isFalse(yield* fs.exists(rotatedProviderLogPath)); + + assert.isTrue(yield* storage.restoreTree(threadId)); + const restoredMessages = yield* sql<{ readonly text: string }>` + SELECT text FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(restoredMessages, [{ text: "keep this conversation" }]); + assert.strictEqual(yield* fs.readFileString(attachmentPath), "image bytes"); + assert.isFalse(yield* fs.exists(providerLogPath)); + + yield* storage.finishRestoreTree(threadId); + const remainingManifestCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.strictEqual(remainingManifestCount[0]?.count, 0); + + yield* storage.deleteThread(threadId); + const remainingShellCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM projection_threads WHERE thread_id = ${threadId} + `; + const remainingMessageCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + assert.strictEqual(remainingShellCount[0]?.count, 0); + assert.strictEqual(remainingMessageCount[0]?.count, 0); + assert.isFalse(yield* fs.exists(attachmentPath)); + + yield* storage.compactLegacyStorage; + const maintenance = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_storage_maintenance + WHERE task = 'compact-legacy-thread-storage' + `; + assert.deepStrictEqual(maintenance, [{ status: "complete" }]); + }), + ); +}); diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts new file mode 100644 index 00000000000..3602e5fb4cd --- /dev/null +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -0,0 +1,491 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeUtil from "node:util"; +import * as NodeZlib from "node:zlib"; + +import { ThreadId } from "@t3tools/contracts"; +import * as Data from "effect/Data"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { parseAttachmentIdFromRelativePath } from "../../attachmentStore.ts"; +import { + parseThreadSegmentFromAttachmentId, + toSafeThreadAttachmentSegment, +} from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; +import { + ThreadColdStorage, + ThreadColdStorageError, + type ThreadColdStorageShape, +} from "../Services/ThreadColdStorage.ts"; + +const gzipAsync = NodeUtil.promisify(NodeZlib.gzip); +const gunzipAsync = NodeUtil.promisify(NodeZlib.gunzip); +const ARCHIVE_SCHEMA = "cold_archive"; +const ARCHIVE_VERSION = 1; +const ROW_CHUNK_SIZE = 250; + +type SqlRow = Record; + +class ArchiveCodecError extends Data.TaggedError("ArchiveCodecError")<{ + readonly cause: unknown; +}> {} + +const THREAD_TABLES = [ + ["orchestration_events", "stream_id"], + ["orchestration_command_receipts", "aggregate_id"], + ["checkpoint_diff_blobs", "thread_id"], + ["provider_session_runtime", "thread_id"], + ["projection_thread_messages", "thread_id"], + ["projection_thread_activities", "thread_id"], + ["projection_thread_sessions", "thread_id"], + ["projection_turns", "thread_id"], + ["projection_pending_approvals", "thread_id"], + ["projection_thread_proposed_plans", "thread_id"], +] as const; + +function storageError(operation: string, threadId: string, cause: unknown) { + return new ThreadColdStorageError({ operation, threadId, cause }); +} + +function encodeRows(rows: ReadonlyArray): Uint8Array { + return Buffer.from(JSON.stringify(rows), "utf8"); +} + +function decodeRows(data: Uint8Array): ReadonlyArray { + return JSON.parse(Buffer.from(data).toString("utf8")) as ReadonlyArray; +} + +const compress = (data: Uint8Array) => + Effect.tryPromise({ + try: () => gzipAsync(data), + catch: (cause) => new ArchiveCodecError({ cause }), + }).pipe(Effect.map((value) => new Uint8Array(value))); + +const decompress = (data: Uint8Array) => + Effect.tryPromise({ + try: () => gunzipAsync(data), + catch: (cause) => new ArchiveCodecError({ cause }), + }).pipe(Effect.map((value) => new Uint8Array(value))); + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig; + + yield* sql.unsafe(`ATTACH DATABASE ? AS ${ARCHIVE_SCHEMA}`, [config.archiveDbPath]); + yield* sql.unsafe(`PRAGMA ${ARCHIVE_SCHEMA}.auto_vacuum = INCREMENTAL`); + yield* sql.unsafe(` + CREATE TABLE IF NOT EXISTS ${ARCHIVE_SCHEMA}.archive_threads ( + thread_id TEXT PRIMARY KEY, + root_thread_id TEXT NOT NULL, + archive_version INTEGER NOT NULL, + archived_at TEXT NOT NULL, + original_bytes INTEGER NOT NULL, + compressed_bytes INTEGER NOT NULL, + created_at TEXT NOT NULL + ) + `); + yield* sql.unsafe(` + CREATE TABLE IF NOT EXISTS ${ARCHIVE_SCHEMA}.archive_thread_chunks ( + thread_id TEXT NOT NULL, + chunk_index INTEGER NOT NULL, + kind TEXT NOT NULL, + row_count INTEGER NOT NULL, + data BLOB NOT NULL, + PRIMARY KEY (thread_id, chunk_index) + ) + `); + yield* sql.unsafe(` + CREATE INDEX IF NOT EXISTS ${ARCHIVE_SCHEMA}.idx_archive_thread_chunks_thread + ON archive_thread_chunks(thread_id, chunk_index) + `); + + const attachmentEntriesForThread = Effect.fn("attachmentEntriesForThread")(function* ( + threadId: string, + ) { + const segment = toSafeThreadAttachmentSegment(threadId); + if (!segment) return [] as string[]; + const entries = yield* fs + .readDirectory(config.attachmentsDir, { recursive: false }) + .pipe(Effect.orElseSucceed(() => [] as string[])); + return entries.filter((entry) => { + const attachmentId = parseAttachmentIdFromRelativePath(entry); + return attachmentId !== null && parseThreadSegmentFromAttachmentId(attachmentId) === segment; + }); + }); + + const removeAttachments = Effect.fn("removeThreadAttachments")(function* (threadId: string) { + const entries = yield* attachmentEntriesForThread(threadId); + yield* Effect.forEach( + entries, + (entry) => fs.remove(path.join(config.attachmentsDir, entry), { force: true }), + { concurrency: 4, discard: true }, + ); + }); + + const removeProviderLogsImpl = Effect.fn("removeProviderLogs")(function* (threadId: string) { + const segment = toSafeThreadAttachmentSegment(threadId); + if (!segment) return; + const baseName = `${segment}.log`; + const entries = yield* fs + .readDirectory(config.providerLogsDir, { recursive: false }) + .pipe(Effect.orElseSucceed(() => [] as string[])); + yield* Effect.forEach( + entries.filter( + (entry) => + entry === baseName || new RegExp(`^${baseName.replace(".", "\\.")}\\.\\d+$`).test(entry), + ), + (entry) => fs.remove(path.join(config.providerLogsDir, entry), { force: true }), + { concurrency: 4, discard: true }, + ); + }); + + const reclaimFreePages = Effect.fn("reclaimThreadStorageFreePages")(function* () { + yield* sql.unsafe("PRAGMA main.incremental_vacuum(2048)"); + yield* sql.unsafe(`PRAGMA ${ARCHIVE_SCHEMA}.incremental_vacuum(2048)`); + }); + + const insertChunk = Effect.fn("insertArchiveChunk")(function* (input: { + readonly threadId: string; + readonly chunkIndex: number; + readonly kind: string; + readonly rowCount: number; + readonly data: Uint8Array; + }) { + yield* sql.unsafe( + `INSERT INTO ${ARCHIVE_SCHEMA}.archive_thread_chunks + (thread_id, chunk_index, kind, row_count, data) + VALUES (?, ?, ?, ?, ?)`, + [input.threadId, input.chunkIndex, input.kind, input.rowCount, input.data], + ); + }); + + const archiveImpl = Effect.fn("archiveThreadImpl")(function* (threadId: ThreadId) { + const manifestRows = (yield* sql.unsafe( + `SELECT root_thread_id, archived_at, status + FROM thread_archive_manifests + WHERE thread_id = ?`, + [threadId], + )) as ReadonlyArray; + const threadRows = (yield* sql.unsafe( + `SELECT thread_id AS root_thread_id, archived_at + FROM projection_threads + WHERE thread_id = ? AND deleted_at IS NULL AND archived_at IS NOT NULL`, + [threadId], + )) as ReadonlyArray; + const source = manifestRows[0] ?? threadRows[0]; + if (!source) return; + if (source.status === "cold") return; + const rootThreadId = String(source.root_thread_id ?? threadId); + const archivedAt = String(source.archived_at ?? DateTime.formatIso(yield* DateTime.now)); + + yield* sql.unsafe( + `INSERT INTO thread_archive_manifests + (thread_id, root_thread_id, status, archive_version, archived_at, updated_at, error) + VALUES (?, ?, 'archiving', ?, ?, CURRENT_TIMESTAMP, NULL) + ON CONFLICT(thread_id) DO UPDATE SET + root_thread_id = excluded.root_thread_id, + status = 'archiving', + archive_version = excluded.archive_version, + archived_at = excluded.archived_at, + updated_at = CURRENT_TIMESTAMP, + error = NULL`, + [threadId, rootThreadId, ARCHIVE_VERSION, archivedAt], + ); + yield* sql.unsafe(`DELETE FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks WHERE thread_id = ?`, [ + threadId, + ]); + yield* sql.unsafe(`DELETE FROM ${ARCHIVE_SCHEMA}.archive_threads WHERE thread_id = ?`, [ + threadId, + ]); + + let chunkIndex = 0; + let originalBytes = 0; + let compressedBytes = 0; + for (const [table, keyColumn] of THREAD_TABLES) { + let lastRowId = 0; + while (true) { + const rows = (yield* sql.unsafe( + `SELECT rowid AS __archive_rowid, * + FROM ${table} + WHERE ${keyColumn} = ? AND rowid > ? + ORDER BY rowid ASC + LIMIT ${ROW_CHUNK_SIZE}`, + [threadId, lastRowId], + )) as ReadonlyArray; + if (rows.length === 0) break; + const normalizedRows = rows.map((row) => { + const { __archive_rowid, ...stored } = row; + lastRowId = Number(__archive_rowid); + return stored; + }); + const encoded = encodeRows(normalizedRows); + const compressed = yield* compress(encoded); + yield* insertChunk({ + threadId, + chunkIndex, + kind: `table:${table}`, + rowCount: normalizedRows.length, + data: compressed, + }); + chunkIndex += 1; + originalBytes += encoded.byteLength; + compressedBytes += compressed.byteLength; + } + } + + const attachmentEntries = yield* attachmentEntriesForThread(threadId); + for (const entry of attachmentEntries) { + const bytes = yield* fs.readFile(path.join(config.attachmentsDir, entry)); + const compressed = yield* compress(bytes); + yield* insertChunk({ + threadId, + chunkIndex, + kind: `attachment:${entry}`, + rowCount: 1, + data: compressed, + }); + chunkIndex += 1; + originalBytes += bytes.byteLength; + compressedBytes += compressed.byteLength; + } + + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql.unsafe( + `INSERT INTO ${ARCHIVE_SCHEMA}.archive_threads + (thread_id, root_thread_id, archive_version, archived_at, original_bytes, + compressed_bytes, created_at) + VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)`, + [threadId, rootThreadId, ARCHIVE_VERSION, archivedAt, originalBytes, compressedBytes], + ); + for (const [table, keyColumn] of [...THREAD_TABLES].toReversed()) { + yield* sql.unsafe(`DELETE FROM ${table} WHERE ${keyColumn} = ?`, [threadId]); + } + yield* sql.unsafe( + `UPDATE thread_archive_manifests + SET status = 'cold', original_bytes = ?, compressed_bytes = ?, + updated_at = CURRENT_TIMESTAMP, error = NULL + WHERE thread_id = ?`, + [originalBytes, compressedBytes, threadId], + ); + }), + ); + + yield* removeAttachments(threadId); + yield* removeProviderLogsImpl(threadId); + yield* reclaimFreePages(); + }); + + const insertRows = Effect.fn("restoreArchiveRows")(function* ( + table: string, + rows: ReadonlyArray, + ) { + for (const row of rows) { + const columns = Object.keys(row); + if (columns.length === 0) continue; + const placeholders = columns.map(() => "?").join(", "); + yield* sql.unsafe( + `INSERT OR REPLACE INTO ${table} (${columns.join(", ")}) VALUES (${placeholders})`, + columns.map((column) => row[column]), + ); + } + }); + + const restoreThread = Effect.fn("restoreArchivedThread")(function* (threadId: ThreadId) { + const chunks = (yield* sql.unsafe( + `SELECT chunk_index, kind, data + FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks + WHERE thread_id = ? + ORDER BY chunk_index ASC`, + [threadId], + )) as ReadonlyArray; + if (chunks.length === 0) return false; + + yield* sql.withTransaction( + Effect.gen(function* () { + for (const chunk of chunks) { + const kind = String(chunk.kind); + const data = yield* decompress(chunk.data as Uint8Array); + if (!kind.startsWith("table:")) continue; + yield* insertRows(kind.slice("table:".length), decodeRows(data)); + } + yield* sql.unsafe( + `UPDATE thread_archive_manifests + SET status = 'restored', updated_at = CURRENT_TIMESTAMP, error = NULL + WHERE thread_id = ?`, + [threadId], + ); + }), + ); + + for (const chunk of chunks) { + const kind = String(chunk.kind); + if (!kind.startsWith("attachment:")) continue; + const entry = kind.slice("attachment:".length); + if (entry.length === 0 || entry.includes("/") || entry.includes("\\")) continue; + const data = yield* decompress(chunk.data as Uint8Array); + yield* fs.writeFile(path.join(config.attachmentsDir, entry), data); + } + return true; + }); + + const resolveTreeRoot = Effect.fn("resolveArchiveTreeRoot")(function* (threadId: ThreadId) { + const rows = (yield* sql.unsafe( + `SELECT COALESCE( + (SELECT root_thread_id FROM thread_archive_manifests WHERE thread_id = ?), + (SELECT thread_id FROM projection_threads WHERE thread_id = ?), + ? + ) AS root_thread_id`, + [threadId, threadId, threadId], + )) as ReadonlyArray; + return ThreadId.make(String(rows[0]?.root_thread_id ?? threadId)); + }); + + const restoreTreeImpl = Effect.fn("restoreArchiveTreeImpl")(function* (threadId: ThreadId) { + const rootThreadId = yield* resolveTreeRoot(threadId); + const rows = (yield* sql.unsafe( + `SELECT thread_id + FROM thread_archive_manifests + WHERE root_thread_id = ? AND status IN ('cold', 'restored') + ORDER BY CASE WHEN thread_id = ? THEN 1 ELSE 0 END, thread_id ASC`, + [rootThreadId, rootThreadId], + )) as ReadonlyArray; + let restored = false; + for (const row of rows) { + restored = (yield* restoreThread(ThreadId.make(String(row.thread_id)))) || restored; + } + return restored; + }); + + const finishRestoreTreeImpl = Effect.fn("finishRestoreArchiveTreeImpl")(function* ( + threadId: ThreadId, + ) { + const rootThreadId = yield* resolveTreeRoot(threadId); + const rows = (yield* sql.unsafe( + `SELECT thread_id FROM thread_archive_manifests WHERE root_thread_id = ? AND status = 'restored'`, + [rootThreadId], + )) as ReadonlyArray; + yield* sql.withTransaction( + Effect.gen(function* () { + for (const row of rows) { + const restoredThreadId = String(row.thread_id); + yield* sql.unsafe( + `DELETE FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks WHERE thread_id = ?`, + [restoredThreadId], + ); + yield* sql.unsafe(`DELETE FROM ${ARCHIVE_SCHEMA}.archive_threads WHERE thread_id = ?`, [ + restoredThreadId, + ]); + yield* sql.unsafe(`DELETE FROM thread_archive_manifests WHERE thread_id = ?`, [ + restoredThreadId, + ]); + } + }), + ); + yield* sql.unsafe(`PRAGMA ${ARCHIVE_SCHEMA}.incremental_vacuum(2048)`); + }); + + const deleteImpl = Effect.fn("deleteThreadPermanentlyImpl")(function* (threadId: ThreadId) { + yield* sql.unsafe( + `INSERT OR IGNORE INTO thread_cleanup_queue (thread_id, reason, created_at) + VALUES (?, 'deleted', CURRENT_TIMESTAMP)`, + [threadId], + ); + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql.unsafe( + `UPDATE projection_thread_proposed_plans + SET implementation_thread_id = NULL + WHERE implementation_thread_id = ?`, + [threadId], + ); + yield* sql.unsafe( + `UPDATE projection_turns + SET source_proposed_plan_thread_id = NULL, source_proposed_plan_id = NULL + WHERE source_proposed_plan_thread_id = ?`, + [threadId], + ); + for (const [table, keyColumn] of [...THREAD_TABLES].toReversed()) { + yield* sql.unsafe(`DELETE FROM ${table} WHERE ${keyColumn} = ?`, [threadId]); + } + yield* sql.unsafe(`DELETE FROM projection_threads WHERE thread_id = ?`, [threadId]); + yield* sql.unsafe( + `DELETE FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks WHERE thread_id = ?`, + [threadId], + ); + yield* sql.unsafe(`DELETE FROM ${ARCHIVE_SCHEMA}.archive_threads WHERE thread_id = ?`, [ + threadId, + ]); + yield* sql.unsafe(`DELETE FROM thread_archive_manifests WHERE thread_id = ?`, [threadId]); + yield* sql.unsafe(`DELETE FROM thread_cleanup_queue WHERE thread_id = ?`, [threadId]); + }), + ); + yield* removeAttachments(threadId); + yield* removeProviderLogsImpl(threadId); + yield* reclaimFreePages(); + }); + + const compactLegacyStorageImpl = Effect.fn("compactLegacyThreadStorage")(function* () { + const rows = (yield* sql.unsafe( + `SELECT status FROM thread_storage_maintenance + WHERE task = 'compact-legacy-thread-storage'`, + )) as ReadonlyArray; + if (rows[0]?.status === "complete") return; + + yield* sql.unsafe( + `UPDATE thread_storage_maintenance + SET status = 'running', updated_at = CURRENT_TIMESTAMP, error = NULL + WHERE task = 'compact-legacy-thread-storage'`, + ); + yield* sql.unsafe("PRAGMA wal_checkpoint(TRUNCATE)"); + yield* sql.unsafe("PRAGMA main.auto_vacuum = INCREMENTAL"); + yield* sql.unsafe("VACUUM main"); + yield* reclaimFreePages(); + yield* sql.unsafe( + `UPDATE thread_storage_maintenance + SET status = 'complete', updated_at = CURRENT_TIMESTAMP, error = NULL + WHERE task = 'compact-legacy-thread-storage'`, + ); + }); + + const wrap = (operation: string, threadId: ThreadId, effect: Effect.Effect) => + effect.pipe(Effect.mapError((cause) => storageError(operation, threadId, cause))); + + const listIds = (query: string, operation: string) => + sql.unsafe(query).pipe( + Effect.map((rows) => + (rows as ReadonlyArray).map((row) => ThreadId.make(String(row.thread_id))), + ), + Effect.mapError((cause) => storageError(operation, "startup", cause)), + ); + + return { + archiveThread: (threadId) => wrap("archive", threadId, archiveImpl(threadId)), + restoreTree: (threadId) => wrap("restore", threadId, restoreTreeImpl(threadId)), + finishRestoreTree: (threadId) => + wrap("finish-restore", threadId, finishRestoreTreeImpl(threadId)), + deleteThread: (threadId) => wrap("delete", threadId, deleteImpl(threadId)), + removeProviderLogs: (threadId) => + wrap("remove-provider-logs", threadId, removeProviderLogsImpl(threadId)), + compactLegacyStorage: compactLegacyStorageImpl().pipe( + Effect.mapError((cause) => storageError("compact-legacy-storage", "startup", cause)), + ), + listPendingArchiveThreadIds: listIds( + `SELECT thread_id FROM thread_archive_manifests WHERE status IN ('pending', 'archiving') ORDER BY archived_at ASC, thread_id ASC`, + "list-pending-archives", + ), + listPendingDeleteThreadIds: listIds( + `SELECT thread_id FROM thread_cleanup_queue WHERE reason = 'deleted' ORDER BY created_at ASC, thread_id ASC`, + "list-pending-deletes", + ), + } satisfies ThreadColdStorageShape; +}); + +export const ThreadColdStorageLive = Layer.effect(ThreadColdStorage, make); diff --git a/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts b/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts index 7d8a24069a3..4023dba8d8d 100644 --- a/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts +++ b/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts @@ -5,15 +5,22 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; +import { ProviderEventLoggers } from "../../provider/Layers/ProviderEventLoggers.ts"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import * as TerminalManager from "../../terminal/Manager.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ThreadColdStorage } from "../Services/ThreadColdStorage.ts"; import { ThreadDeletionReactor, type ThreadDeletionReactorShape, } from "../Services/ThreadDeletionReactor.ts"; type ThreadDeletedEvent = Extract; +type ThreadArchivedEvent = Extract; +type ThreadLifecycleJob = + | { readonly type: "archive"; readonly threadId: ThreadArchivedEvent["payload"]["threadId"] } + | { readonly type: "delete"; readonly threadId: ThreadDeletedEvent["payload"]["threadId"] } + | { readonly type: "compact-legacy-storage" }; export const logCleanupCauseUnlessInterrupted = ({ effect, @@ -40,6 +47,8 @@ const make = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const providerService = yield* ProviderService; const terminalManager = yield* TerminalManager.TerminalManager; + const threadColdStorage = yield* ThreadColdStorage; + const providerEventLoggers = yield* ProviderEventLoggers; const stopProviderSession = (threadId: ThreadDeletedEvent["payload"]["threadId"]) => logCleanupCauseUnlessInterrupted({ @@ -55,39 +64,88 @@ const make = Effect.gen(function* () { threadId, }); - const processThreadDeleted = Effect.fn("processThreadDeleted")(function* ( - event: ThreadDeletedEvent, + const closeProviderLogWriters = (threadId: ThreadDeletedEvent["payload"]["threadId"]) => + logCleanupCauseUnlessInterrupted({ + effect: Effect.all( + [providerEventLoggers.native, providerEventLoggers.canonical].flatMap((logger) => + logger?.closeThread ? [logger.closeThread(threadId)] : [], + ), + { discard: true }, + ), + message: "thread lifecycle cleanup skipped provider log writer close", + threadId, + }); + + const processLifecycleJob = Effect.fn("processThreadLifecycleJob")(function* ( + job: ThreadLifecycleJob, ) { - const { threadId } = event.payload; + if (job.type === "compact-legacy-storage") { + yield* threadColdStorage.compactLegacyStorage; + return; + } + const { threadId } = job; yield* stopProviderSession(threadId); yield* closeThreadTerminals(threadId); + yield* closeProviderLogWriters(threadId); + if (job.type === "archive") { + yield* threadColdStorage.archiveThread(threadId); + } else { + yield* threadColdStorage.deleteThread(threadId); + } }); - const processThreadDeletedSafely = (event: ThreadDeletedEvent) => - processThreadDeleted(event).pipe( + const processLifecycleJobSafely = (job: ThreadLifecycleJob) => + processLifecycleJob(job).pipe( Effect.catchCause((cause) => { if (Cause.hasInterruptsOnly(cause)) { return Effect.failCause(cause); } - return Effect.logWarning("thread deletion reactor failed to process event", { - eventType: event.type, - threadId: event.payload.threadId, + return Effect.logWarning("thread lifecycle reactor failed to process job", { + lifecycleAction: job.type, + ...(job.type === "compact-legacy-storage" ? {} : { threadId: job.threadId }), cause: Cause.pretty(cause), }); }), ); - const worker = yield* makeDrainableWorker(processThreadDeletedSafely); + const worker = yield* makeDrainableWorker(processLifecycleJobSafely); const start: ThreadDeletionReactorShape["start"] = Effect.fn("start")(function* () { yield* Effect.forkScoped( Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { - if (event.type !== "thread.deleted") { - return Effect.void; + if (event.type === "thread.deleted") { + return worker.enqueue({ type: "delete", threadId: event.payload.threadId }); + } + if (event.type === "thread.archived") { + return worker.enqueue({ type: "archive", threadId: event.payload.threadId }); } - return worker.enqueue(event); + return Effect.void; }), ); + + const pendingJobs = yield* Effect.all([ + threadColdStorage.listPendingDeleteThreadIds, + threadColdStorage.listPendingArchiveThreadIds, + ]).pipe( + Effect.catchCause((cause) => + Effect.logWarning("failed to read pending thread storage migrations", { + cause: Cause.pretty(cause), + }).pipe(Effect.as(null)), + ), + ); + if (pendingJobs === null) return; + const [pendingDeletes, pendingArchives] = pendingJobs; + yield* Effect.forEach( + pendingDeletes, + (threadId) => worker.enqueue({ type: "delete", threadId }), + { discard: true }, + ); + yield* Effect.forEach( + pendingArchives, + (threadId) => worker.enqueue({ type: "archive", threadId }), + { discard: true }, + ); + yield* worker.enqueue({ type: "compact-legacy-storage" }); }); return { diff --git a/apps/server/src/orchestration/Services/ThreadColdStorage.ts b/apps/server/src/orchestration/Services/ThreadColdStorage.ts new file mode 100644 index 00000000000..17380914ac6 --- /dev/null +++ b/apps/server/src/orchestration/Services/ThreadColdStorage.ts @@ -0,0 +1,38 @@ +import type { ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +export class ThreadColdStorageError extends Schema.TaggedErrorClass()( + "ThreadColdStorageError", + { + operation: Schema.String, + threadId: Schema.String, + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + return `Cold thread storage failed during ${this.operation} for '${this.threadId}'.`; + } +} + +export interface ThreadColdStorageShape { + readonly archiveThread: (threadId: ThreadId) => Effect.Effect; + readonly restoreTree: (threadId: ThreadId) => Effect.Effect; + readonly finishRestoreTree: (threadId: ThreadId) => Effect.Effect; + readonly deleteThread: (threadId: ThreadId) => Effect.Effect; + readonly removeProviderLogs: (threadId: ThreadId) => Effect.Effect; + readonly compactLegacyStorage: Effect.Effect; + readonly listPendingArchiveThreadIds: Effect.Effect< + ReadonlyArray, + ThreadColdStorageError + >; + readonly listPendingDeleteThreadIds: Effect.Effect< + ReadonlyArray, + ThreadColdStorageError + >; +} + +export class ThreadColdStorage extends Context.Service()( + "t3/orchestration/Services/ThreadColdStorage", +) {} diff --git a/apps/server/src/orchestration/Services/ThreadDeletionReactor.ts b/apps/server/src/orchestration/Services/ThreadDeletionReactor.ts index 7c6718965a6..7159e0fa72a 100644 --- a/apps/server/src/orchestration/Services/ThreadDeletionReactor.ts +++ b/apps/server/src/orchestration/Services/ThreadDeletionReactor.ts @@ -1,8 +1,8 @@ /** * ThreadDeletionReactor - Thread deletion cleanup reactor service interface. * - * Owns background workers that react to thread deletion domain events and - * perform best-effort runtime cleanup for provider sessions and terminals. + * Owns background workers that react to thread archive/delete domain events + * and perform runtime cleanup plus durable cold-storage lifecycle work. * * @module ThreadDeletionReactor */ diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index ba1131ee259..244aa9d00dd 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -45,6 +45,8 @@ import Migration0029 from "./Migrations/029_ProjectionThreadDetailOrderingIndexe import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes.ts"; import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; +import Migration0033 from "./Migrations/033_ThreadColdArchive.ts"; +import Migration0034 from "./Migrations/034_DeletedThreadCleanupQueue.ts"; /** * Migration loader with all migrations defined inline. @@ -89,6 +91,8 @@ export const migrationEntries = [ [30, "ProjectionThreadShellArchiveIndexes", Migration0030], [31, "AuthAuthorizationScopes", Migration0031], [32, "AuthPairingProofKeyThumbprint", Migration0032], + [33, "ThreadColdArchive", Migration0033], + [34, "DeletedThreadCleanupQueue", Migration0034], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/033_034_ThreadStorageLifecycle.test.ts b/apps/server/src/persistence/Migrations/033_034_ThreadStorageLifecycle.test.ts new file mode 100644 index 00000000000..17619226a70 --- /dev/null +++ b/apps/server/src/persistence/Migrations/033_034_ThreadStorageLifecycle.test.ts @@ -0,0 +1,69 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("033-034 thread storage lifecycle migrations", (it) => { + it.effect("queues archived and deleted threads", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 32 }); + + const insertThread = (input: { + readonly threadId: string; + readonly archivedAt: string | null; + readonly deletedAt: string | null; + }) => sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, + interaction_mode, created_at, updated_at, archived_at, deleted_at + ) VALUES ( + ${input.threadId}, 'project-1', ${input.threadId}, + '{"instanceId":"codex","model":"gpt-5.5","options":[]}', + 'full-access', 'default', '2026-07-01T00:00:00.000Z', + '2026-07-01T00:00:00.000Z', ${input.archivedAt}, ${input.deletedAt} + ) + `; + + yield* insertThread({ + threadId: "archived-thread", + archivedAt: "2026-07-02T00:00:00.000Z", + deletedAt: null, + }); + yield* insertThread({ + threadId: "deleted-thread", + archivedAt: null, + deletedAt: "2026-07-03T00:00:00.000Z", + }); + + yield* runMigrations({ toMigrationInclusive: 34 }); + + const manifests = yield* sql<{ + readonly threadId: string; + readonly rootThreadId: string; + readonly status: string; + }>` + SELECT thread_id AS "threadId", root_thread_id AS "rootThreadId", status + FROM thread_archive_manifests + ORDER BY thread_id + `; + assert.deepStrictEqual(manifests, [ + { + threadId: "archived-thread", + rootThreadId: "archived-thread", + status: "pending", + }, + ]); + + const cleanup = yield* sql<{ readonly threadId: string; readonly reason: string }>` + SELECT thread_id AS "threadId", reason FROM thread_cleanup_queue + `; + assert.deepStrictEqual(cleanup, [{ threadId: "deleted-thread", reason: "deleted" }]); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/033_ThreadColdArchive.ts b/apps/server/src/persistence/Migrations/033_ThreadColdArchive.ts new file mode 100644 index 00000000000..d10e7e5394b --- /dev/null +++ b/apps/server/src/persistence/Migrations/033_ThreadColdArchive.ts @@ -0,0 +1,68 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +/** + * Registers already-archived threads for conversion to cold storage. + * + * The migration deliberately only creates durable work records. Compression + * and filesystem I/O are performed by the background lifecycle worker after + * startup so a large existing archive cannot hold the schema migration or UI + * thread hostage. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS thread_archive_manifests ( + thread_id TEXT PRIMARY KEY, + root_thread_id TEXT NOT NULL, + status TEXT NOT NULL, + archive_version INTEGER NOT NULL DEFAULT 1, + archived_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + original_bytes INTEGER NOT NULL DEFAULT 0, + compressed_bytes INTEGER NOT NULL DEFAULT 0, + error TEXT + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_thread_archive_manifests_root_status + ON thread_archive_manifests(root_thread_id, status, archived_at, thread_id) + `; + + yield* sql` + CREATE TABLE IF NOT EXISTS thread_storage_maintenance ( + task TEXT PRIMARY KEY, + status TEXT NOT NULL, + updated_at TEXT NOT NULL, + error TEXT + ) + `; + + yield* sql` + INSERT OR IGNORE INTO thread_storage_maintenance (task, status, updated_at) + VALUES ('compact-legacy-thread-storage', 'pending', CURRENT_TIMESTAMP) + `; + + yield* sql` + INSERT OR IGNORE INTO thread_archive_manifests ( + thread_id, + root_thread_id, + status, + archive_version, + archived_at, + updated_at + ) + SELECT + thread_id, + thread_id, + 'pending', + 1, + archived_at, + CURRENT_TIMESTAMP + FROM projection_threads + WHERE archived_at IS NOT NULL + AND deleted_at IS NULL + `; +}); diff --git a/apps/server/src/persistence/Migrations/034_DeletedThreadCleanupQueue.ts b/apps/server/src/persistence/Migrations/034_DeletedThreadCleanupQueue.ts new file mode 100644 index 00000000000..24cc80ec44e --- /dev/null +++ b/apps/server/src/persistence/Migrations/034_DeletedThreadCleanupQueue.ts @@ -0,0 +1,24 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +/** Queues legacy soft-deleted threads for the same permanent cleanup as new deletes. */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS thread_cleanup_queue ( + thread_id TEXT PRIMARY KEY, + reason TEXT NOT NULL, + created_at TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT + ) + `; + + yield* sql` + INSERT OR IGNORE INTO thread_cleanup_queue (thread_id, reason, created_at) + SELECT thread_id, 'deleted', CURRENT_TIMESTAMP + FROM projection_threads + WHERE deleted_at IS NOT NULL + `; +}); diff --git a/apps/server/src/provider/Layers/EventNdjsonLogger.ts b/apps/server/src/provider/Layers/EventNdjsonLogger.ts index 8c20a4c1936..a4b80d78cf5 100644 --- a/apps/server/src/provider/Layers/EventNdjsonLogger.ts +++ b/apps/server/src/provider/Layers/EventNdjsonLogger.ts @@ -33,6 +33,7 @@ export type EventNdjsonStream = "native" | "canonical" | "orchestration"; export interface EventNdjsonLogger { readonly filePath: string; write: (event: unknown, threadId: ThreadId | null) => Effect.Effect; + closeThread?: (threadId: ThreadId) => Effect.Effect; close: () => Effect.Effect; } @@ -278,9 +279,27 @@ export const makeEventNdjsonLogger = Effect.fn("makeEventNdjsonLogger")(function ); }); + const closeThread = Effect.fn("closeThread")(function* (threadId: ThreadId) { + const threadSegment = resolveThreadSegment(threadId); + yield* SynchronizedRef.modifyEffect(stateRef, (state) => { + const writer = state.threadWriters.get(threadSegment); + if (!writer) { + return Effect.succeed([undefined, state] as const); + } + return writer.close().pipe( + Effect.map(() => { + const nextThreadWriters = new Map(state.threadWriters); + nextThreadWriters.delete(threadSegment); + return [undefined, { ...state, threadWriters: nextThreadWriters }] as const; + }), + ); + }); + }); + return { filePath, write, + closeThread, close, } satisfies EventNdjsonLogger; }); diff --git a/apps/server/src/provider/acp/AcpNativeLogging.test.ts b/apps/server/src/provider/acp/AcpNativeLogging.test.ts index 8c92d523aee..b2904cc7913 100644 --- a/apps/server/src/provider/acp/AcpNativeLogging.test.ts +++ b/apps/server/src/provider/acp/AcpNativeLogging.test.ts @@ -21,6 +21,7 @@ nodeServicesIt("ACP native logging", (it) => { const nativeEventLogger: EventNdjsonLogger = { filePath: "/tmp/provider-native.ndjson", write: (event) => Effect.sync(() => void records.push(event)), + closeThread: () => Effect.void, close: () => Effect.void, }; const makeLogger = yield* makeAcpNativeLoggerFactory(); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 0c632d8486c..f3455a32a03 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -49,6 +49,7 @@ import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRun import { ProviderCommandReactorLive } from "./orchestration/Layers/ProviderCommandReactor.ts"; import { CheckpointReactorLive } from "./orchestration/Layers/CheckpointReactor.ts"; import { ThreadDeletionReactorLive } from "./orchestration/Layers/ThreadDeletionReactor.ts"; +import { ThreadColdStorageLive } from "./orchestration/Layers/ThreadColdStorage.ts"; import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts"; import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry.ts"; @@ -279,9 +280,13 @@ const CloudManagedEndpointRuntimeLive = Layer.mergeAll( ), ); +const OrchestrationLayerWithColdStorageLive = OrchestrationLayerLive.pipe( + Layer.provideMerge(ThreadColdStorageLive), +); + const ProviderRuntimeLayerLive = ProviderSessionReaperLive.pipe( Layer.provideMerge(ProviderLayerLive), - Layer.provideMerge(OrchestrationLayerLive), + Layer.provideMerge(OrchestrationLayerWithColdStorageLive), ); const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index bb6b2752a17..2053c6a7287 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -92,6 +92,7 @@ import { useThreadDiscoveredPorts } from "../portDiscoveryState"; import { openDiscoveredPort } from "./preview/openDiscoveredPort"; import { useAtomCommand } from "../state/use-atom-command"; import { previewEnvironment } from "../state/preview"; +import { useOptimisticThreadArchiveStore } from "../optimisticThreadArchiveStore"; import { legacyProjectCwdPreferenceKey, resolveProjectExpanded, @@ -3107,6 +3108,9 @@ const SidebarProjectsContent = memo(function SidebarProjectsContent( export default function Sidebar() { const projects = useProjects(); const sidebarThreads = useThreadShells(); + const optimisticallyArchivedThreadKeys = useOptimisticThreadArchiveStore( + (state) => state.threadKeys, + ); const projectExpandedById = useUiStateStore((store) => store.projectExpandedById); const projectOrder = useUiStateStore((store) => store.projectOrder); const reorderProjects = useUiStateStore((store) => store.reorderProjects); @@ -3368,8 +3372,15 @@ export default function Sidebar() { }, []); const visibleThreads = useMemo( - () => sidebarThreads.filter((thread) => thread.archivedAt === null), - [sidebarThreads], + () => + sidebarThreads.filter( + (thread) => + thread.archivedAt === null && + !optimisticallyArchivedThreadKeys.has( + scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), + ), + ), + [optimisticallyArchivedThreadKeys, sidebarThreads], ); const sortedProjects = useMemo(() => { const sortableProjects = sidebarProjects.map((project) => ({ diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index 40017d56314..ab258959a73 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -11,7 +11,7 @@ import { type ProviderInstanceId, type ScopedThreadRef, } from "@t3tools/contracts"; -import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; import { safeErrorLogAttributes } from "@t3tools/client-runtime/errors"; import { isAtomCommandInterrupted, @@ -1423,6 +1423,9 @@ export function ProviderSettingsPanel() { export function ArchivedThreadsPanel() { const projects = useProjects(); const { unarchiveThread, confirmAndDeleteThread } = useThreadActions(); + const [unarchivingThreadKeys, setUnarchivingThreadKeys] = useState>( + () => new Set(), + ); const environmentIds = useMemo( () => [...new Set(projects.map((project) => project.environmentId))], [projects], @@ -1484,19 +1487,11 @@ export function ArchivedThreadsPanel() { return groups; }, [archivedSnapshots]); - const handleArchivedThreadContextMenu = useCallback( - async (threadRef: ScopedThreadRef, position: { x: number; y: number }) => { - const api = readLocalApi(); - if (!api) return; - const clicked = await api.contextMenu.show( - [ - { id: "unarchive", label: "Unarchive" }, - { id: "delete", label: "Delete", destructive: true }, - ], - position, - ); - - if (clicked === "unarchive") { + const handleUnarchiveThread = useCallback( + async (threadRef: ScopedThreadRef) => { + const threadKey = scopedThreadKey(threadRef); + setUnarchivingThreadKeys((current) => new Set(current).add(threadKey)); + try { const result = await unarchiveThread(threadRef); if (result._tag === "Success") { refreshArchivedThreads(); @@ -1510,6 +1505,31 @@ export function ArchivedThreadsPanel() { }), ); } + } finally { + setUnarchivingThreadKeys((current) => { + const next = new Set(current); + next.delete(threadKey); + return next; + }); + } + }, + [refreshArchivedThreads, unarchiveThread], + ); + + const handleArchivedThreadContextMenu = useCallback( + async (threadRef: ScopedThreadRef, position: { x: number; y: number }) => { + const api = readLocalApi(); + if (!api) return; + const clicked = await api.contextMenu.show( + [ + { id: "unarchive", label: "Unarchive" }, + { id: "delete", label: "Delete", destructive: true }, + ], + position, + ); + + if (clicked === "unarchive") { + await handleUnarchiveThread(threadRef); return; } @@ -1529,7 +1549,7 @@ export function ArchivedThreadsPanel() { } } }, - [confirmAndDeleteThread, refreshArchivedThreads, unarchiveThread], + [confirmAndDeleteThread, handleUnarchiveThread, refreshArchivedThreads], ); return ( @@ -1565,11 +1585,15 @@ export function ArchivedThreadsPanel() { title={project.name} icon={} > - {projectThreads.map((thread) => ( - { + const threadRef = scopeThreadRef(thread.environmentId, thread.id); + const isUnarchiving = unarchivingThreadKeys.has(scopedThreadKey(threadRef)); + return ( + { event.preventDefault(); + if (isUnarchiving) return; void (async () => { const result = await settlePromise(() => handleArchivedThreadContextMenu( @@ -1607,35 +1631,22 @@ export function ArchivedThreadsPanel() { variant="outline" size="sm" className="h-7 shrink-0 cursor-pointer gap-1.5 px-2.5" + disabled={isUnarchiving} onClick={() => { - void (async () => { - const result = await unarchiveThread( - scopeThreadRef(thread.environmentId, thread.id), - ); - if (result._tag === "Success") { - refreshArchivedThreads(); - return; - } - if (!isAtomCommandInterrupted(result)) { - const error = squashAtomCommandFailure(result); - toastManager.add( - stackedThreadToast({ - type: "error", - title: "Failed to unarchive thread", - description: - error instanceof Error ? error.message : "An error occurred.", - }), - ); - } - })(); + void handleUnarchiveThread(threadRef); }} > - - Unarchive + {isUnarchiving ? ( + + ) : ( + + )} + {isUnarchiving ? "Unarchiving" : "Unarchive"} } /> - ))} + ); + })} )) )} diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 07655ad30d7..ed049106b45 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -26,6 +26,10 @@ import { formatWorktreePathForDisplay, getOrphanedWorktreePathForThread } from " import { stackedThreadToast, toastManager } from "../components/ui/toast"; import { useClientSettings } from "./useSettings"; import { useAtomCommand } from "../state/use-atom-command"; +import { + optimisticallyHideArchivedThread, + revealOptimisticallyArchivedThread, +} from "../optimisticThreadArchiveStore"; export class ThreadArchiveBlockedError extends Schema.TaggedErrorClass()( "ThreadArchiveBlockedError", @@ -108,25 +112,31 @@ export function useThreadActions() { const shouldNavigateToDraft = currentRouteThreadRef?.threadId === threadRef.threadId && currentRouteThreadRef.environmentId === threadRef.environmentId; - const archiveResult = await archiveThreadMutation({ + optimisticallyHideArchivedThread(threadRef); + const archivePromise = archiveThreadMutation({ environmentId: threadRef.environmentId, input: { threadId: threadRef.threadId }, }); - if (archiveResult._tag === "Failure") { - return archiveResult; - } - if (shouldNavigateToDraft) { const navigationResult = await settlePromise(() => handleNewThreadRef.current(scopeProjectRef(thread.environmentId, thread.projectId)), ); if (navigationResult._tag === "Failure") { + revealOptimisticallyArchivedThread(threadRef); return navigationResult; } - refreshArchivedThreadsForEnvironment(threadRef.environmentId); + } + + const archiveResult = await archivePromise; + if (archiveResult._tag === "Failure") { + revealOptimisticallyArchivedThread(threadRef); return archiveResult; } + // The domain event is published before the command acknowledgement, so + // the shell now owns visibility. Do not retain a local tombstone that + // could hide a later unarchive performed by another client. + revealOptimisticallyArchivedThread(threadRef); refreshArchivedThreadsForEnvironment(threadRef.environmentId); return archiveResult; }, @@ -135,6 +145,7 @@ export function useThreadActions() { const unarchiveThread = useCallback( async (target: ScopedThreadRef) => { + revealOptimisticallyArchivedThread(target); const result = await unarchiveThreadMutation({ environmentId: target.environmentId, input: { threadId: target.threadId }, diff --git a/apps/web/src/optimisticThreadArchiveStore.ts b/apps/web/src/optimisticThreadArchiveStore.ts new file mode 100644 index 00000000000..b9a1951c9f6 --- /dev/null +++ b/apps/web/src/optimisticThreadArchiveStore.ts @@ -0,0 +1,33 @@ +import type { ScopedThreadRef } from "@t3tools/contracts"; +import { scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { create } from "zustand"; + +interface OptimisticThreadArchiveState { + readonly threadKeys: ReadonlySet; + readonly hide: (threadRef: ScopedThreadRef) => void; + readonly show: (threadRef: ScopedThreadRef) => void; +} + +export const useOptimisticThreadArchiveStore = create((set) => ({ + threadKeys: new Set(), + hide: (threadRef) => + set((state) => { + const next = new Set(state.threadKeys); + next.add(scopedThreadKey(threadRef)); + return { threadKeys: next }; + }), + show: (threadRef) => + set((state) => { + const next = new Set(state.threadKeys); + next.delete(scopedThreadKey(threadRef)); + return { threadKeys: next }; + }), +})); + +export function optimisticallyHideArchivedThread(threadRef: ScopedThreadRef): void { + useOptimisticThreadArchiveStore.getState().hide(threadRef); +} + +export function revealOptimisticallyArchivedThread(threadRef: ScopedThreadRef): void { + useOptimisticThreadArchiveStore.getState().show(threadRef); +} From b56f8f54dfa1dae559b206ec345fcafee8a31741 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Tue, 14 Jul 2026 13:00:55 +0100 Subject: [PATCH 02/10] Harden archived thread lifecycle recovery --- .../archive/ArchivedThreadsScreen.tsx | 4 +- .../Layers/OrchestrationEngine.test.ts | 118 +++++++++++++++ .../Layers/OrchestrationEngine.ts | 32 ++++ .../Layers/ThreadColdStorage.test.ts | 137 ++++++++++++++++++ .../orchestration/Layers/ThreadColdStorage.ts | 56 ++++++- .../Services/ThreadColdStorage.ts | 1 + apps/web/src/hooks/useThreadActions.ts | 5 + 7 files changed, 345 insertions(+), 8 deletions(-) diff --git a/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx b/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx index a36d39b73ea..7f455002c96 100644 --- a/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx +++ b/apps/mobile/src/features/archive/ArchivedThreadsScreen.tsx @@ -402,6 +402,7 @@ function ArchivedThreadRow(props: { const subtitle = [props.environmentLabel, props.thread.branch].filter((part): part is string => Boolean(part), ); + const onDelete = props.isUnarchiving ? () => undefined : props.onDelete; return ( ProjectId.make(value); const asMessageId = (value: string): MessageId => MessageId.make(value); @@ -864,6 +866,122 @@ describe("OrchestrationEngine", () => { await runtime.dispose(); }); + effectIt.effect("rolls restored cold storage back when unarchive dispatch fails", () => { + type StoredEvent = + ReturnType extends Effect.Effect + ? A + : never; + const events: StoredEvent[] = []; + const rolledBackThreadIds: ThreadId[] = []; + const finishedThreadIds: ThreadId[] = []; + let nextSequence = 1; + let restoreOnUnarchive = false; + + const flakyStore: OrchestrationEventStoreShape = { + append(event) { + if (event.commandId === CommandId.make("cmd-cold-unarchive-fail")) { + return Effect.fail( + new PersistenceSqlError({ + operation: "test.append", + detail: "unarchive append failed", + }), + ); + } + const savedEvent = { ...event, sequence: nextSequence } as StoredEvent; + nextSequence += 1; + events.push(savedEvent); + return Effect.succeed(savedEvent); + }, + readFromSequence(sequenceExclusive) { + return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); + }, + readAll() { + return Stream.fromIterable(events); + }, + }; + const coldStorage: ThreadColdStorageShape = { + archiveThread: () => Effect.void, + restoreTree: () => Effect.succeed(restoreOnUnarchive), + rollbackRestoreTree: (threadId) => + Effect.sync(() => { + rolledBackThreadIds.push(threadId); + }), + finishRestoreTree: (threadId) => + Effect.sync(() => { + finishedThreadIds.push(threadId); + }), + deleteThread: () => Effect.void, + removeProviderLogs: () => Effect.void, + compactLegacyStorage: Effect.void, + listPendingArchiveThreadIds: Effect.succeed([]), + listPendingDeleteThreadIds: Effect.succeed([]), + }; + const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { + prefix: "t3-orchestration-engine-test-", + }); + const testLayer = OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(OrchestrationProjectionPipelineLive), + Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)), + Layer.provide(Layer.succeed(ThreadColdStorage, coldStorage)), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provide(RepositoryIdentityResolver.layer), + Layer.provide(SqlitePersistenceMemory), + Layer.provideMerge(ServerConfigLayer), + Layer.provideMerge(NodeServices.layer), + ); + return Effect.gen(function* () { + const engine = yield* OrchestrationEngineService; + const createdAt = now(); + + yield* engine.dispatch({ + type: "project.create", + commandId: CommandId.make("cmd-cold-project-create"), + projectId: asProjectId("project-cold-rollback"), + title: "Cold rollback project", + workspaceRoot: "/tmp/project-cold-rollback", + defaultModelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + createdAt, + }); + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-cold-thread-create"), + threadId: ThreadId.make("thread-cold-rollback"), + projectId: asProjectId("project-cold-rollback"), + title: "Cold rollback thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt, + }); + yield* engine.dispatch({ + type: "thread.archive", + commandId: CommandId.make("cmd-cold-thread-archive"), + threadId: ThreadId.make("thread-cold-rollback"), + }); + restoreOnUnarchive = true; + + const unarchiveFailure = yield* Effect.flip( + engine.dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("cmd-cold-unarchive-fail"), + threadId: ThreadId.make("thread-cold-rollback"), + }), + ); + expect(unarchiveFailure.message).toContain("unarchive append failed"); + expect(rolledBackThreadIds).toEqual([ThreadId.make("thread-cold-rollback")]); + expect(finishedThreadIds).toEqual([]); + }).pipe(Effect.provide(testLayer)); + }); + it("rolls back all events for a multi-event command when projection fails mid-dispatch", async () => { let shouldFailRequestedProjection = true; const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = { diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 6fab829441e..9e323e929ee 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -107,6 +107,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => { const dispatchStartSequence = commandReadModel.snapshotSequence; let processingStartedAtMs = 0; + let restoredUnarchiveThreadId: ThreadId | null = null; const aggregateRef = commandToAggregateRef(envelope.command); const baseMetricAttributes = { commandType: envelope.command.type, @@ -166,6 +167,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { ), ); if (restored) { + restoredUnarchiveThreadId = unarchiveThreadId; commandReadModel = yield* projectionSnapshotQuery.getCommandReadModel(); } } @@ -291,6 +293,36 @@ const makeOrchestrationEngine = Effect.gen(function* () { return; } + if (restoredUnarchiveThreadId !== null && Option.isSome(threadColdStorage)) { + const rolledBack = yield* threadColdStorage.value + .rollbackRestoreTree(restoredUnarchiveThreadId) + .pipe( + Effect.as(true), + Effect.catchCause((cause) => + Effect.logWarning("failed to roll back restored archive bundle", { + threadId: restoredUnarchiveThreadId, + cause: Cause.pretty(cause), + }).pipe(Effect.as(false)), + ), + ); + if (rolledBack) { + const refreshedReadModel = yield* projectionSnapshotQuery.getCommandReadModel().pipe( + Effect.catchCause((cause) => + Effect.logWarning( + "failed to refresh orchestration read model after archive rollback", + { + threadId: restoredUnarchiveThreadId, + cause: Cause.pretty(cause), + }, + ).pipe(Effect.as(null)), + ), + ); + if (refreshedReadModel !== null) { + commandReadModel = refreshedReadModel; + } + } + } + const error = Cause.squash(exit.cause) as OrchestrationDispatchError; if (!isOrchestrationCommandPreviouslyRejectedError(error)) { yield* reconcileReadModelAfterDispatchFailure.pipe( diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts index 15f16175c63..528ec182954 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts @@ -12,6 +12,26 @@ import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { ThreadColdStorage } from "../Services/ThreadColdStorage.ts"; import { ThreadColdStorageLive } from "./ThreadColdStorage.ts"; +const insertArchivedThread = Effect.fn("insertArchivedThreadTestFixture")(function* ( + threadId: ThreadId, + title: string, +) { + const sql = yield* SqlClient.SqlClient; + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, + interaction_mode, created_at, updated_at, archived_at, parent_kind, + root_thread_id + ) VALUES ( + ${threadId}, 'project-1', ${title}, + '{"instanceId":"codex","model":"gpt-5.5","options":[]}', + 'full-access', 'default', '2026-07-01T00:00:00.000Z', + '2026-07-02T00:00:00.000Z', '2026-07-02T00:00:00.000Z', + 'root', ${threadId} + ) + `; +}); + const layer = it.layer( ThreadColdStorageLive.pipe( Layer.provideMerge(SqlitePersistenceMemory), @@ -103,6 +123,19 @@ layer("ThreadColdStorage", (it) => { assert.strictEqual(yield* fs.readFileString(attachmentPath), "image bytes"); assert.isFalse(yield* fs.exists(providerLogPath)); + yield* storage.rollbackRestoreTree(threadId); + const rolledBackMessages = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + const rolledBackManifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(rolledBackMessages, [{ count: 0 }]); + assert.deepStrictEqual(rolledBackManifest, [{ status: "cold" }]); + + assert.isTrue(yield* storage.restoreTree(threadId)); + assert.strictEqual(yield* fs.readFileString(attachmentPath), "image bytes"); + yield* storage.finishRestoreTree(threadId); const remainingManifestCount = yield* sql<{ readonly count: number }>` SELECT COUNT(*) AS count FROM thread_archive_manifests WHERE thread_id = ${threadId} @@ -128,4 +161,108 @@ layer("ThreadColdStorage", (it) => { assert.deepStrictEqual(maintenance, [{ status: "complete" }]); }), ); + + it.effect("ignores traversal attachment entries while restoring", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const threadId = ThreadId.make("thread-traversal"); + const attachmentName = "thread-traversal-00000000-0000-4000-8000-000000000001.png"; + + yield* insertArchivedThread(threadId, "Traversal thread"); + yield* fs.writeFileString(path.join(config.attachmentsDir, attachmentName), "image bytes"); + yield* storage.archiveThread(threadId); + yield* sql` + UPDATE cold_archive.archive_thread_chunks + SET kind = 'attachment:..' + WHERE thread_id = ${threadId} AND kind LIKE 'attachment:%' + `; + + assert.isTrue(yield* storage.restoreTree(threadId)); + const manifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(manifest, [{ status: "restored" }]); + }), + ); + + it.effect("retries archive cleanup without rebuilding deleted hot data", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const threadId = ThreadId.make("thread-cleanup-retry"); + const providerLogPath = path.join(config.providerLogsDir, "thread-cleanup-retry.log"); + + yield* insertArchivedThread(threadId, "Cleanup retry thread"); + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES ( + 'message-cleanup-retry', ${threadId}, NULL, 'user', 'preserve across cleanup retry', + '[]', 0, '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z' + ) + `; + yield* fs.makeDirectory(providerLogPath); + yield* fs.writeFileString(path.join(providerLogPath, "keep"), "force cleanup failure"); + + const archiveFailure = yield* Effect.flip(storage.archiveThread(threadId)); + assert.strictEqual(archiveFailure.operation, "archive"); + const pendingManifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(pendingManifest, [{ status: "cleanup_pending" }]); + assert.deepInclude(yield* storage.listPendingArchiveThreadIds, threadId); + + yield* fs.remove(providerLogPath, { recursive: true }); + yield* storage.archiveThread(threadId); + const coldManifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(coldManifest, [{ status: "cold" }]); + + assert.isTrue(yield* storage.restoreTree(threadId)); + const restoredMessages = yield* sql<{ readonly text: string }>` + SELECT text FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(restoredMessages, [{ text: "preserve across cleanup retry" }]); + }), + ); + + it.effect("keeps the delete cleanup queue entry until external cleanup succeeds", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const threadId = ThreadId.make("thread-delete-retry"); + const attachmentName = "thread-delete-retry-00000000-0000-4000-8000-000000000001.png"; + const attachmentPath = path.join(config.attachmentsDir, attachmentName); + + yield* insertArchivedThread(threadId, "Delete retry thread"); + yield* fs.makeDirectory(attachmentPath); + yield* fs.writeFileString(path.join(attachmentPath, "keep"), "force cleanup failure"); + + const deleteFailure = yield* Effect.flip(storage.deleteThread(threadId)); + assert.strictEqual(deleteFailure.operation, "delete"); + const pendingCleanup = yield* sql<{ readonly reason: string }>` + SELECT reason FROM thread_cleanup_queue WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(pendingCleanup, [{ reason: "deleted" }]); + + yield* fs.remove(attachmentPath, { recursive: true }); + yield* storage.deleteThread(threadId); + const completedCleanup = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM thread_cleanup_queue WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(completedCleanup, [{ count: 0 }]); + }), + ); }); diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts index 3602e5fb4cd..0ee4b626bf9 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -151,6 +151,20 @@ const make = Effect.gen(function* () { yield* sql.unsafe(`PRAGMA ${ARCHIVE_SCHEMA}.incremental_vacuum(2048)`); }); + const completeArchiveCleanup = Effect.fn("completeThreadArchiveCleanup")(function* ( + threadId: ThreadId, + ) { + yield* removeAttachments(threadId); + yield* removeProviderLogsImpl(threadId); + yield* reclaimFreePages(); + yield* sql.unsafe( + `UPDATE thread_archive_manifests + SET status = 'cold', updated_at = CURRENT_TIMESTAMP, error = NULL + WHERE thread_id = ? AND status = 'cleanup_pending'`, + [threadId], + ); + }); + const insertChunk = Effect.fn("insertArchiveChunk")(function* (input: { readonly threadId: string; readonly chunkIndex: number; @@ -182,6 +196,10 @@ const make = Effect.gen(function* () { const source = manifestRows[0] ?? threadRows[0]; if (!source) return; if (source.status === "cold") return; + if (source.status === "cleanup_pending") { + yield* completeArchiveCleanup(threadId); + return; + } const rootThreadId = String(source.root_thread_id ?? threadId); const archivedAt = String(source.archived_at ?? DateTime.formatIso(yield* DateTime.now)); @@ -270,7 +288,7 @@ const make = Effect.gen(function* () { } yield* sql.unsafe( `UPDATE thread_archive_manifests - SET status = 'cold', original_bytes = ?, compressed_bytes = ?, + SET status = 'cleanup_pending', original_bytes = ?, compressed_bytes = ?, updated_at = CURRENT_TIMESTAMP, error = NULL WHERE thread_id = ?`, [originalBytes, compressedBytes, threadId], @@ -278,9 +296,7 @@ const make = Effect.gen(function* () { }), ); - yield* removeAttachments(threadId); - yield* removeProviderLogsImpl(threadId); - yield* reclaimFreePages(); + yield* completeArchiveCleanup(threadId); }); const insertRows = Effect.fn("restoreArchiveRows")(function* ( @@ -329,7 +345,15 @@ const make = Effect.gen(function* () { const kind = String(chunk.kind); if (!kind.startsWith("attachment:")) continue; const entry = kind.slice("attachment:".length); - if (entry.length === 0 || entry.includes("/") || entry.includes("\\")) continue; + if ( + entry.length === 0 || + entry === "." || + entry === ".." || + entry.includes("/") || + entry.includes("\\") + ) { + continue; + } const data = yield* decompress(chunk.data as Uint8Array); yield* fs.writeFile(path.join(config.attachmentsDir, entry), data); } @@ -364,6 +388,22 @@ const make = Effect.gen(function* () { return restored; }); + const rollbackRestoreTreeImpl = Effect.fn("rollbackRestoreArchiveTreeImpl")(function* ( + threadId: ThreadId, + ) { + const rootThreadId = yield* resolveTreeRoot(threadId); + const rows = (yield* sql.unsafe( + `SELECT thread_id + FROM thread_archive_manifests + WHERE root_thread_id = ? AND status = 'restored' + ORDER BY CASE WHEN thread_id = ? THEN 1 ELSE 0 END, thread_id ASC`, + [rootThreadId, rootThreadId], + )) as ReadonlyArray; + for (const row of rows) { + yield* archiveImpl(ThreadId.make(String(row.thread_id))); + } + }); + const finishRestoreTreeImpl = Effect.fn("finishRestoreArchiveTreeImpl")(function* ( threadId: ThreadId, ) { @@ -424,12 +464,12 @@ const make = Effect.gen(function* () { threadId, ]); yield* sql.unsafe(`DELETE FROM thread_archive_manifests WHERE thread_id = ?`, [threadId]); - yield* sql.unsafe(`DELETE FROM thread_cleanup_queue WHERE thread_id = ?`, [threadId]); }), ); yield* removeAttachments(threadId); yield* removeProviderLogsImpl(threadId); yield* reclaimFreePages(); + yield* sql.unsafe(`DELETE FROM thread_cleanup_queue WHERE thread_id = ?`, [threadId]); }); const compactLegacyStorageImpl = Effect.fn("compactLegacyThreadStorage")(function* () { @@ -469,6 +509,8 @@ const make = Effect.gen(function* () { return { archiveThread: (threadId) => wrap("archive", threadId, archiveImpl(threadId)), restoreTree: (threadId) => wrap("restore", threadId, restoreTreeImpl(threadId)), + rollbackRestoreTree: (threadId) => + wrap("rollback-restore", threadId, rollbackRestoreTreeImpl(threadId)), finishRestoreTree: (threadId) => wrap("finish-restore", threadId, finishRestoreTreeImpl(threadId)), deleteThread: (threadId) => wrap("delete", threadId, deleteImpl(threadId)), @@ -478,7 +520,7 @@ const make = Effect.gen(function* () { Effect.mapError((cause) => storageError("compact-legacy-storage", "startup", cause)), ), listPendingArchiveThreadIds: listIds( - `SELECT thread_id FROM thread_archive_manifests WHERE status IN ('pending', 'archiving') ORDER BY archived_at ASC, thread_id ASC`, + `SELECT thread_id FROM thread_archive_manifests WHERE status IN ('pending', 'archiving', 'cleanup_pending') ORDER BY archived_at ASC, thread_id ASC`, "list-pending-archives", ), listPendingDeleteThreadIds: listIds( diff --git a/apps/server/src/orchestration/Services/ThreadColdStorage.ts b/apps/server/src/orchestration/Services/ThreadColdStorage.ts index 17380914ac6..acf8fe76f98 100644 --- a/apps/server/src/orchestration/Services/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Services/ThreadColdStorage.ts @@ -19,6 +19,7 @@ export class ThreadColdStorageError extends Schema.TaggedErrorClass Effect.Effect; readonly restoreTree: (threadId: ThreadId) => Effect.Effect; + readonly rollbackRestoreTree: (threadId: ThreadId) => Effect.Effect; readonly finishRestoreTree: (threadId: ThreadId) => Effect.Effect; readonly deleteThread: (threadId: ThreadId) => Effect.Effect; readonly removeProviderLogs: (threadId: ThreadId) => Effect.Effect; diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index ed049106b45..0eca6109d06 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -123,6 +123,11 @@ export function useThreadActions() { ); if (navigationResult._tag === "Failure") { revealOptimisticallyArchivedThread(threadRef); + void archivePromise.then((archiveResult) => { + if (archiveResult._tag === "Success") { + refreshArchivedThreadsForEnvironment(threadRef.environmentId); + } + }); return navigationResult; } } From 1b6c373a8a0784e64c0ba57504383fbe163290f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Tue, 14 Jul 2026 13:17:29 +0100 Subject: [PATCH 03/10] Harden cold archive restoration - Preserve binary SQL values across archive round trips - Keep cold bundles authoritative until attachments restore safely - Bound restore memory and tolerate compatible schema changes --- .../Layers/OrchestrationEngine.ts | 6 +- .../Layers/ThreadColdStorage.test.ts | 85 ++++++- .../orchestration/Layers/ThreadColdStorage.ts | 214 ++++++++++++++---- .../components/settings/SettingsPanels.tsx | 106 ++++----- 4 files changed, 311 insertions(+), 100 deletions(-) diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 9e323e929ee..25f462ceada 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -250,11 +250,11 @@ const makeOrchestrationEngine = Effect.gen(function* () { ); } } - if (unarchiveThreadId !== null && Option.isSome(threadColdStorage)) { - yield* threadColdStorage.value.finishRestoreTree(unarchiveThreadId).pipe( + if (restoredUnarchiveThreadId !== null && Option.isSome(threadColdStorage)) { + yield* threadColdStorage.value.finishRestoreTree(restoredUnarchiveThreadId).pipe( Effect.catchCause((cause) => Effect.logWarning("failed to finalize restored archive bundle", { - threadId: unarchiveThreadId, + threadId: restoredUnarchiveThreadId, cause: Cause.pretty(cause), }), ), diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts index 528ec182954..142f460d993 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts @@ -20,14 +20,12 @@ const insertArchivedThread = Effect.fn("insertArchivedThreadTestFixture")(functi yield* sql` INSERT INTO projection_threads ( thread_id, project_id, title, model_selection_json, runtime_mode, - interaction_mode, created_at, updated_at, archived_at, parent_kind, - root_thread_id + interaction_mode, created_at, updated_at, archived_at ) VALUES ( ${threadId}, 'project-1', ${title}, '{"instanceId":"codex","model":"gpt-5.5","options":[]}', 'full-access', 'default', '2026-07-01T00:00:00.000Z', - '2026-07-02T00:00:00.000Z', '2026-07-02T00:00:00.000Z', - 'root', ${threadId} + '2026-07-02T00:00:00.000Z', '2026-07-02T00:00:00.000Z' ) `; }); @@ -175,9 +173,10 @@ layer("ThreadColdStorage", (it) => { yield* insertArchivedThread(threadId, "Traversal thread"); yield* fs.writeFileString(path.join(config.attachmentsDir, attachmentName), "image bytes"); yield* storage.archiveThread(threadId); + const escapedPath = path.join(config.attachmentsDir, "..", "thread-traversal-escape"); yield* sql` UPDATE cold_archive.archive_thread_chunks - SET kind = 'attachment:..' + SET kind = 'attachment:../thread-traversal-escape' WHERE thread_id = ${threadId} AND kind LIKE 'attachment:%' `; @@ -186,6 +185,82 @@ layer("ThreadColdStorage", (it) => { SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} `; assert.deepStrictEqual(manifest, [{ status: "restored" }]); + assert.isFalse(yield* fs.exists(escapedPath)); + }), + ); + + it.effect("keeps cold SQL data authoritative when attachment restore fails", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const threadId = ThreadId.make("thread-attachment-restore-failure"); + const attachmentName = + "thread-attachment-restore-failure-00000000-0000-4000-8000-000000000001.png"; + + yield* insertArchivedThread(threadId, "Attachment restore failure"); + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES ( + 'message-attachment-restore-failure', ${threadId}, NULL, 'user', 'keep me cold', + '[]', 0, '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z' + ) + `; + yield* fs.writeFileString(path.join(config.attachmentsDir, attachmentName), "image bytes"); + yield* storage.archiveThread(threadId); + + const blockedTarget = path.join(config.attachmentsDir, "blocked-restore.bin"); + yield* fs.makeDirectory(blockedTarget); + yield* fs.writeFileString(path.join(blockedTarget, "keep"), "prevent replacement"); + yield* sql` + UPDATE cold_archive.archive_thread_chunks + SET kind = 'attachment:blocked-restore.bin' + WHERE thread_id = ${threadId} AND kind LIKE 'attachment:%' + `; + + const failure = yield* Effect.flip(storage.restoreTree(threadId)); + assert.strictEqual(failure.operation, "restore"); + const messages = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + const manifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + const chunks = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM cold_archive.archive_thread_chunks WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(messages, [{ count: 0 }]); + assert.deepStrictEqual(manifest, [{ status: "cold" }]); + assert.isAbove(chunks[0]?.count ?? 0, 0); + }), + ); + + it.effect("round-trips binary SQL values", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const threadId = ThreadId.make("thread-binary"); + const diffBytes = new Uint8Array([0, 1, 2, 127, 128, 255]); + + yield* insertArchivedThread(threadId, "Binary thread"); + yield* sql.unsafe( + `INSERT INTO checkpoint_diff_blobs + (thread_id, from_turn_count, to_turn_count, diff, created_at) + VALUES (?, 0, 1, ?, '2026-07-01T00:00:00.000Z')`, + [threadId, diffBytes], + ); + + yield* storage.archiveThread(threadId); + assert.isTrue(yield* storage.restoreTree(threadId)); + const restored = (yield* sql.unsafe( + `SELECT diff FROM checkpoint_diff_blobs WHERE thread_id = ?`, + [threadId], + )) as ReadonlyArray<{ readonly diff: Uint8Array }>; + assert.deepStrictEqual(restored[0]?.diff, diffBytes); }), ); diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts index 0ee4b626bf9..be180ac11eb 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -9,10 +9,11 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; -import { parseAttachmentIdFromRelativePath } from "../../attachmentStore.ts"; import { + parseAttachmentIdFromRelativePath, parseThreadSegmentFromAttachmentId, toSafeThreadAttachmentSegment, } from "../../attachmentStore.ts"; @@ -28,6 +29,10 @@ const gunzipAsync = NodeUtil.promisify(NodeZlib.gunzip); const ARCHIVE_SCHEMA = "cold_archive"; const ARCHIVE_VERSION = 1; const ROW_CHUNK_SIZE = 250; +const RESTORE_CHUNK_PAGE_SIZE = 32; +const BINARY_VALUE_KEY = "__t3_archive_binary_base64"; +const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.UnknownFromJsonString); +const decodeUnknownJsonString = Schema.decodeUnknownEffect(Schema.UnknownFromJsonString); type SqlRow = Record; @@ -52,12 +57,84 @@ function storageError(operation: string, threadId: string, cause: unknown) { return new ThreadColdStorageError({ operation, threadId, cause }); } -function encodeRows(rows: ReadonlyArray): Uint8Array { - return Buffer.from(JSON.stringify(rows), "utf8"); +function encodeRows(rows: ReadonlyArray): Effect.Effect { + return encodeUnknownJsonString( + rows.map((row) => + Object.fromEntries( + Object.entries(row).map(([column, value]) => [ + column, + value instanceof Uint8Array + ? { [BINARY_VALUE_KEY]: Buffer.from(value).toString("base64") } + : value, + ]), + ), + ), + ).pipe( + Effect.map((encoded) => Buffer.from(encoded, "utf8")), + Effect.mapError((cause) => new ArchiveCodecError({ cause })), + ); +} + +function decodeRows(data: Uint8Array): Effect.Effect, ArchiveCodecError> { + return decodeUnknownJsonString(Buffer.from(data).toString("utf8")).pipe( + Effect.flatMap((decoded) => + Effect.try({ + try: () => { + if (!Array.isArray(decoded)) { + throw new TypeError("Archived table chunk must contain an array of rows"); + } + return decoded.map((row): SqlRow => { + if (row === null || typeof row !== "object" || Array.isArray(row)) { + throw new TypeError("Archived table chunk contains an invalid row"); + } + return Object.fromEntries( + Object.entries(row).map(([column, value]) => { + if ( + value !== null && + typeof value === "object" && + !Array.isArray(value) && + Object.keys(value).length === 1 && + typeof (value as Record)[BINARY_VALUE_KEY] === "string" + ) { + return [ + column, + new Uint8Array( + Buffer.from( + (value as Record)[BINARY_VALUE_KEY] as string, + "base64", + ), + ), + ]; + } + return [column, value]; + }), + ); + }); + }, + catch: (cause) => new ArchiveCodecError({ cause }), + }), + ), + Effect.mapError((cause) => + cause instanceof ArchiveCodecError ? cause : new ArchiveCodecError({ cause }), + ), + ); } -function decodeRows(data: Uint8Array): ReadonlyArray { - return JSON.parse(Buffer.from(data).toString("utf8")) as ReadonlyArray; +const THREAD_TABLE_NAMES = new Set(THREAD_TABLES.map(([table]) => table)); + +function quoteIdentifier(identifier: string): string { + return `"${identifier.replaceAll('"', '""')}"`; +} + +function isSafeAttachmentEntry(entry: string): boolean { + return ( + entry.length > 0 && + entry !== "." && + entry !== ".." && + !entry.includes("/") && + !entry.includes("\\") && + !entry.includes("\0") + ); } const compress = (data: Uint8Array) => @@ -137,10 +214,15 @@ const make = Effect.gen(function* () { .readDirectory(config.providerLogsDir, { recursive: false }) .pipe(Effect.orElseSucceed(() => [] as string[])); yield* Effect.forEach( - entries.filter( - (entry) => - entry === baseName || new RegExp(`^${baseName.replace(".", "\\.")}\\.\\d+$`).test(entry), - ), + entries.filter((entry) => { + if (entry === baseName) return true; + if (!entry.startsWith(`${baseName}.`)) return false; + const rotation = entry.slice(baseName.length + 1); + return ( + rotation.length > 0 && + [...rotation].every((character) => character >= "0" && character <= "9") + ); + }), (entry) => fs.remove(path.join(config.providerLogsDir, entry), { force: true }), { concurrency: 4, discard: true }, ); @@ -243,7 +325,7 @@ const make = Effect.gen(function* () { lastRowId = Number(__archive_rowid); return stored; }); - const encoded = encodeRows(normalizedRows); + const encoded = yield* encodeRows(normalizedRows); const compressed = yield* compress(encoded); yield* insertChunk({ threadId, @@ -274,6 +356,8 @@ const make = Effect.gen(function* () { compressedBytes += compressed.byteLength; } + // Chunk creation stays retryable outside the hot-row deletion transaction. + // A retry replaces every partial chunk before deleting source data. yield* sql.withTransaction( Effect.gen(function* () { yield* sql.unsafe( @@ -303,34 +387,102 @@ const make = Effect.gen(function* () { table: string, rows: ReadonlyArray, ) { + if (!THREAD_TABLE_NAMES.has(table)) { + return yield* new ArchiveCodecError({ + cause: new TypeError(`Archive chunk targets unknown table '${table}'`), + }); + } + const tableInfo = (yield* sql.unsafe( + `PRAGMA main.table_info(${quoteIdentifier(table)})`, + )) as ReadonlyArray; + const currentColumns = new Set( + tableInfo.flatMap((column) => + typeof column.name === "string" && column.name.length > 0 ? [column.name] : [], + ), + ); for (const row of rows) { - const columns = Object.keys(row); + // Archived rows can outlive schema migrations. Ignore columns that no + // longer exist and let newly-added columns use their database defaults. + const columns = Object.keys(row).filter((column) => currentColumns.has(column)); if (columns.length === 0) continue; const placeholders = columns.map(() => "?").join(", "); yield* sql.unsafe( - `INSERT OR REPLACE INTO ${table} (${columns.join(", ")}) VALUES (${placeholders})`, + `INSERT OR REPLACE INTO ${quoteIdentifier(table)} (${columns.map(quoteIdentifier).join(", ")}) VALUES (${placeholders})`, columns.map((column) => row[column]), ); } }); const restoreThread = Effect.fn("restoreArchivedThread")(function* (threadId: ThreadId) { - const chunks = (yield* sql.unsafe( - `SELECT chunk_index, kind, data - FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks - WHERE thread_id = ? - ORDER BY chunk_index ASC`, + const bundleRows = (yield* sql.unsafe( + `SELECT 1 AS present FROM ${ARCHIVE_SCHEMA}.archive_threads WHERE thread_id = ? LIMIT 1`, [threadId], )) as ReadonlyArray; - if (chunks.length === 0) return false; + if (bundleRows.length === 0) return false; + + const invalidKinds = (yield* sql.unsafe( + `SELECT kind FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks + WHERE thread_id = ? AND kind NOT LIKE 'table:%' AND kind NOT LIKE 'attachment:%' + LIMIT 1`, + [threadId], + )) as ReadonlyArray; + if (invalidKinds.length > 0) { + return yield* new ArchiveCodecError({ + cause: new TypeError( + `Archive contains unknown chunk kind '${String(invalidKinds[0]?.kind)}'`, + ), + }); + } + + // Restore files before changing SQL state. A failed or interrupted file + // write leaves the cold bundle authoritative and can be retried safely. + let attachmentChunkIndex = -1; + while (true) { + const attachmentChunks = (yield* sql.unsafe( + `SELECT chunk_index, kind, data + FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks + WHERE thread_id = ? AND chunk_index > ? AND kind LIKE 'attachment:%' + ORDER BY chunk_index ASC + LIMIT ${RESTORE_CHUNK_PAGE_SIZE}`, + [threadId, attachmentChunkIndex], + )) as ReadonlyArray; + if (attachmentChunks.length === 0) break; + for (const chunk of attachmentChunks) { + attachmentChunkIndex = Number(chunk.chunk_index); + const entry = String(chunk.kind).slice("attachment:".length); + if (!isSafeAttachmentEntry(entry)) continue; + const data = yield* decompress(chunk.data as Uint8Array); + const targetPath = path.join(config.attachmentsDir, entry); + const temporaryPath = `${targetPath}.t3-restore`; + yield* fs + .writeFile(temporaryPath, data) + .pipe( + Effect.andThen(fs.remove(targetPath, { force: true })), + Effect.andThen(fs.rename(temporaryPath, targetPath)), + Effect.ensuring(fs.remove(temporaryPath, { force: true }).pipe(Effect.ignore)), + ); + } + } yield* sql.withTransaction( Effect.gen(function* () { - for (const chunk of chunks) { - const kind = String(chunk.kind); - const data = yield* decompress(chunk.data as Uint8Array); - if (!kind.startsWith("table:")) continue; - yield* insertRows(kind.slice("table:".length), decodeRows(data)); + let tableChunkIndex = -1; + while (true) { + const tableChunks = (yield* sql.unsafe( + `SELECT chunk_index, kind, data + FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks + WHERE thread_id = ? AND chunk_index > ? AND kind LIKE 'table:%' + ORDER BY chunk_index ASC + LIMIT ${RESTORE_CHUNK_PAGE_SIZE}`, + [threadId, tableChunkIndex], + )) as ReadonlyArray; + if (tableChunks.length === 0) break; + for (const chunk of tableChunks) { + tableChunkIndex = Number(chunk.chunk_index); + const kind = String(chunk.kind); + const data = yield* decompress(chunk.data as Uint8Array); + yield* insertRows(kind.slice("table:".length), yield* decodeRows(data)); + } } yield* sql.unsafe( `UPDATE thread_archive_manifests @@ -341,22 +493,6 @@ const make = Effect.gen(function* () { }), ); - for (const chunk of chunks) { - const kind = String(chunk.kind); - if (!kind.startsWith("attachment:")) continue; - const entry = kind.slice("attachment:".length); - if ( - entry.length === 0 || - entry === "." || - entry === ".." || - entry.includes("/") || - entry.includes("\\") - ) { - continue; - } - const data = yield* decompress(chunk.data as Uint8Array); - yield* fs.writeFile(path.join(config.attachmentsDir, entry), data); - } return true; }); diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index ab258959a73..95ec76b694a 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -1591,60 +1591,60 @@ export function ArchivedThreadsPanel() { return ( { - event.preventDefault(); - if (isUnarchiving) return; - void (async () => { - const result = await settlePromise(() => - handleArchivedThreadContextMenu( - scopeThreadRef(thread.environmentId, thread.id), - { - x: event.clientX, - y: event.clientY, - }, - ), - ); - if (result._tag === "Failure") { - const error = squashAtomCommandFailure(result); - toastManager.add( - stackedThreadToast({ - type: "error", - title: "Archived thread action failed", - description: - error instanceof Error ? error.message : "An error occurred.", - }), + onContextMenu={(event) => { + event.preventDefault(); + if (isUnarchiving) return; + void (async () => { + const result = await settlePromise(() => + handleArchivedThreadContextMenu( + scopeThreadRef(thread.environmentId, thread.id), + { + x: event.clientX, + y: event.clientY, + }, + ), ); - } - })(); - }} - title={thread.title} - description={ - <> - Archived {formatRelativeTimeLabel(thread.archivedAt ?? thread.createdAt)} - {" \u00b7 Created "} - {formatRelativeTimeLabel(thread.createdAt)} - - } - control={ - - } - /> + if (result._tag === "Failure") { + const error = squashAtomCommandFailure(result); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Archived thread action failed", + description: + error instanceof Error ? error.message : "An error occurred.", + }), + ); + } + })(); + }} + title={thread.title} + description={ + <> + Archived {formatRelativeTimeLabel(thread.archivedAt ?? thread.createdAt)} + {" \u00b7 Created "} + {formatRelativeTimeLabel(thread.createdAt)} + + } + control={ + + } + /> ); })} From 71f28adc812f366bd339dfec94c3d649bbb8d3a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Wed, 15 Jul 2026 21:54:57 +0100 Subject: [PATCH 04/10] Harden cold storage lifecycle races - Serialize archive-tree lifecycle and recheck archived shells - Restore cleanup-pending bundles before unarchive commits - Preserve retry state for writer and filesystem failures --- .../Layers/OrchestrationEngine.ts | 28 ++-- .../Layers/ThreadColdStorage.test.ts | 140 ++++++++++++++++++ .../orchestration/Layers/ThreadColdStorage.ts | 124 ++++++++++++++-- .../Layers/ThreadDeletionReactor.ts | 31 ++-- 4 files changed, 291 insertions(+), 32 deletions(-) diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 25f462ceada..6e1defce881 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -108,6 +108,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { const dispatchStartSequence = commandReadModel.snapshotSequence; let processingStartedAtMs = 0; let restoredUnarchiveThreadId: ThreadId | null = null; + let restoredUnarchiveCommitted = false; const aggregateRef = commandToAggregateRef(envelope.command); const baseMetricAttributes = { commandType: envelope.command.type, @@ -234,7 +235,18 @@ const makeOrchestrationEngine = Effect.gen(function* () { ), ); + restoredUnarchiveCommitted = restoredUnarchiveThreadId !== null; commandReadModel = committedCommand.nextCommandReadModel; + if (restoredUnarchiveThreadId !== null && Option.isSome(threadColdStorage)) { + yield* threadColdStorage.value.finishRestoreTree(restoredUnarchiveThreadId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("failed to finalize restored archive bundle", { + threadId: restoredUnarchiveThreadId, + cause: Cause.pretty(cause), + }), + ), + ); + } for (const [index, event] of committedCommand.committedEvents.entries()) { yield* PubSub.publish(eventPubSub, event); if (index === 0) { @@ -250,16 +262,6 @@ const makeOrchestrationEngine = Effect.gen(function* () { ); } } - if (restoredUnarchiveThreadId !== null && Option.isSome(threadColdStorage)) { - yield* threadColdStorage.value.finishRestoreTree(restoredUnarchiveThreadId).pipe( - Effect.catchCause((cause) => - Effect.logWarning("failed to finalize restored archive bundle", { - threadId: restoredUnarchiveThreadId, - cause: Cause.pretty(cause), - }), - ), - ); - } return { sequence: committedCommand.lastSequence }; }).pipe(Effect.withSpan(`orchestration.command.${envelope.command.type}`)), ).pipe( @@ -293,7 +295,11 @@ const makeOrchestrationEngine = Effect.gen(function* () { return; } - if (restoredUnarchiveThreadId !== null && Option.isSome(threadColdStorage)) { + if ( + restoredUnarchiveThreadId !== null && + !restoredUnarchiveCommitted && + Option.isSome(threadColdStorage) + ) { const rolledBack = yield* threadColdStorage.value .rollbackRestoreTree(restoredUnarchiveThreadId) .pipe( diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts index 142f460d993..75d556e8909 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts @@ -39,6 +39,17 @@ const layer = it.layer( ); layer("ThreadColdStorage", (it) => { + it.effect("discovers archived shells before a lifecycle manifest exists", () => + Effect.gen(function* () { + const storage = yield* ThreadColdStorage; + const threadId = ThreadId.make("thread-archive-queue-fallback"); + + yield* insertArchivedThread(threadId, "Archive queue fallback thread"); + + assert.deepInclude(yield* storage.listPendingArchiveThreadIds, threadId); + }), + ); + it.effect("compresses conversation data, destroys logs, restores content, and hard-deletes", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -121,6 +132,18 @@ layer("ThreadColdStorage", (it) => { assert.strictEqual(yield* fs.readFileString(attachmentPath), "image bytes"); assert.isFalse(yield* fs.exists(providerLogPath)); + // A queued archive job can run after restore but before the unarchive + // command commits. It must not undo a restore owned by that command. + yield* storage.archiveThread(threadId); + const stillRestoredMessages = yield* sql<{ readonly text: string }>` + SELECT text FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + const stillRestoredManifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(stillRestoredMessages, [{ text: "keep this conversation" }]); + assert.deepStrictEqual(stillRestoredManifest, [{ status: "restored" }]); + yield* storage.rollbackRestoreTree(threadId); const rolledBackMessages = yield* sql<{ readonly count: number }>` SELECT COUNT(*) AS count FROM projection_thread_messages WHERE thread_id = ${threadId} @@ -310,6 +333,123 @@ layer("ThreadColdStorage", (it) => { }), ); + it.effect("restores cleanup-pending bundles before unarchiving", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const threadId = ThreadId.make("thread-cleanup-pending-restore"); + + yield* insertArchivedThread(threadId, "Cleanup-pending restore thread"); + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES ( + 'message-cleanup-pending-restore', ${threadId}, NULL, 'user', + 'restore while cleanup is pending', '[]', 0, + '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z' + ) + `; + + yield* storage.archiveThread(threadId); + yield* sql` + UPDATE thread_archive_manifests + SET status = 'cleanup_pending' + WHERE thread_id = ${threadId} + `; + + assert.isTrue(yield* storage.restoreTree(threadId)); + const restoredMessages = yield* sql<{ readonly text: string }>` + SELECT text FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + const manifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(restoredMessages, [{ text: "restore while cleanup is pending" }]); + assert.deepStrictEqual(manifest, [{ status: "restored" }]); + }), + ); + + it.effect("abandons an incomplete archive after the shell is unarchived", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const threadId = ThreadId.make("thread-unarchived-before-archive"); + + yield* insertArchivedThread(threadId, "Unarchived before archive thread"); + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES ( + 'message-unarchived-before-archive', ${threadId}, NULL, 'user', + 'keep active data hot', '[]', 0, + '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z' + ) + `; + yield* sql` + INSERT INTO thread_archive_manifests ( + thread_id, root_thread_id, status, archive_version, archived_at, updated_at + ) VALUES ( + ${threadId}, ${threadId}, 'pending', 1, + '2026-07-02T00:00:00.000Z', CURRENT_TIMESTAMP + ) + `; + yield* sql` + UPDATE projection_threads SET archived_at = NULL WHERE thread_id = ${threadId} + `; + + yield* storage.archiveThread(threadId); + const messages = yield* sql<{ readonly text: string }>` + SELECT text FROM projection_thread_messages WHERE thread_id = ${threadId} + `; + const manifests = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + const chunks = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM cold_archive.archive_thread_chunks + WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(messages, [{ text: "keep active data hot" }]); + assert.deepStrictEqual(manifests, [{ count: 0 }]); + assert.deepStrictEqual(chunks, [{ count: 0 }]); + }), + ); + + it.effect("retries attachment cleanup after directory I/O errors", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const threadId = ThreadId.make("thread-attachment-directory-error"); + + yield* insertArchivedThread(threadId, "Attachment directory error thread"); + yield* fs.remove(config.attachmentsDir, { recursive: true }); + yield* fs.writeFileString(config.attachmentsDir, "not a directory"); + + const archiveFailure = yield* Effect.flip(storage.archiveThread(threadId)); + assert.strictEqual(archiveFailure.operation, "archive"); + const shell = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM projection_threads WHERE thread_id = ${threadId} + `; + const manifests = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(shell, [{ count: 1 }]); + assert.deepStrictEqual(manifests, [{ status: "archiving" }]); + + yield* fs.remove(config.attachmentsDir); + yield* fs.makeDirectory(config.attachmentsDir); + yield* storage.archiveThread(threadId); + const completedManifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(completedManifest, [{ status: "cold" }]); + }), + ); + it.effect("keeps the delete cleanup queue entry until external cleanup succeeds", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts index be180ac11eb..e430707010e 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -8,8 +8,11 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; +import * as SynchronizedRef from "effect/SynchronizedRef"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { @@ -154,6 +157,7 @@ const make = Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const config = yield* ServerConfig; + const threadLocksRef = yield* SynchronizedRef.make(new Map()); yield* sql.unsafe(`ATTACH DATABASE ? AS ${ARCHIVE_SCHEMA}`, [config.archiveDbPath]); yield* sql.unsafe(`PRAGMA ${ARCHIVE_SCHEMA}.auto_vacuum = INCREMENTAL`); @@ -190,7 +194,11 @@ const make = Effect.gen(function* () { if (!segment) return [] as string[]; const entries = yield* fs .readDirectory(config.attachmentsDir, { recursive: false }) - .pipe(Effect.orElseSucceed(() => [] as string[])); + .pipe( + Effect.catch((error) => + error.reason._tag === "NotFound" ? Effect.succeed([] as string[]) : Effect.fail(error), + ), + ); return entries.filter((entry) => { const attachmentId = parseAttachmentIdFromRelativePath(entry); return attachmentId !== null && parseThreadSegmentFromAttachmentId(attachmentId) === segment; @@ -212,7 +220,11 @@ const make = Effect.gen(function* () { const baseName = `${segment}.log`; const entries = yield* fs .readDirectory(config.providerLogsDir, { recursive: false }) - .pipe(Effect.orElseSucceed(() => [] as string[])); + .pipe( + Effect.catch((error) => + error.reason._tag === "NotFound" ? Effect.succeed([] as string[]) : Effect.fail(error), + ), + ); yield* Effect.forEach( entries.filter((entry) => { if (entry === baseName) return true; @@ -262,7 +274,31 @@ const make = Effect.gen(function* () { ); }); - const archiveImpl = Effect.fn("archiveThreadImpl")(function* (threadId: ThreadId) { + const discardIncompleteArchive = Effect.fn("discardIncompleteThreadArchive")(function* ( + threadId: ThreadId, + ) { + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql.unsafe( + `DELETE FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks WHERE thread_id = ?`, + [threadId], + ); + yield* sql.unsafe(`DELETE FROM ${ARCHIVE_SCHEMA}.archive_threads WHERE thread_id = ?`, [ + threadId, + ]); + yield* sql.unsafe( + `DELETE FROM thread_archive_manifests + WHERE thread_id = ? AND status IN ('pending', 'archiving')`, + [threadId], + ); + }), + ); + }); + + const archiveImpl = Effect.fn("archiveThreadImpl")(function* ( + threadId: ThreadId, + allowRestored: boolean, + ) { const manifestRows = (yield* sql.unsafe( `SELECT root_thread_id, archived_at, status FROM thread_archive_manifests @@ -275,13 +311,21 @@ const make = Effect.gen(function* () { WHERE thread_id = ? AND deleted_at IS NULL AND archived_at IS NOT NULL`, [threadId], )) as ReadonlyArray; - const source = manifestRows[0] ?? threadRows[0]; + const manifest = manifestRows[0]; + const source = manifest ?? threadRows[0]; if (!source) return; + if (threadRows.length === 0) { + if (source.status === "pending" || source.status === "archiving") { + yield* discardIncompleteArchive(threadId); + } + return; + } if (source.status === "cold") return; if (source.status === "cleanup_pending") { yield* completeArchiveCleanup(threadId); return; } + if (source.status === "restored" && !allowRestored) return; const rootThreadId = String(source.root_thread_id ?? threadId); const archivedAt = String(source.archived_at ?? DateTime.formatIso(yield* DateTime.now)); @@ -358,8 +402,30 @@ const make = Effect.gen(function* () { // Chunk creation stays retryable outside the hot-row deletion transaction. // A retry replaces every partial chunk before deleting source data. - yield* sql.withTransaction( + const archivedAtDestructiveBoundary = yield* sql.withTransaction( Effect.gen(function* () { + const archivedShell = (yield* sql.unsafe( + `SELECT 1 AS present + FROM projection_threads + WHERE thread_id = ? AND deleted_at IS NULL AND archived_at IS NOT NULL + LIMIT 1`, + [threadId], + )) as ReadonlyArray; + if (archivedShell.length === 0) { + yield* sql.unsafe( + `DELETE FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks WHERE thread_id = ?`, + [threadId], + ); + yield* sql.unsafe(`DELETE FROM ${ARCHIVE_SCHEMA}.archive_threads WHERE thread_id = ?`, [ + threadId, + ]); + yield* sql.unsafe( + `DELETE FROM thread_archive_manifests + WHERE thread_id = ? AND status = 'archiving'`, + [threadId], + ); + return false; + } yield* sql.unsafe( `INSERT INTO ${ARCHIVE_SCHEMA}.archive_threads (thread_id, root_thread_id, archive_version, archived_at, original_bytes, @@ -377,9 +443,12 @@ const make = Effect.gen(function* () { WHERE thread_id = ?`, [originalBytes, compressedBytes, threadId], ); + return true; }), ); + if (!archivedAtDestructiveBoundary) return; + yield* completeArchiveCleanup(threadId); }); @@ -513,7 +582,7 @@ const make = Effect.gen(function* () { const rows = (yield* sql.unsafe( `SELECT thread_id FROM thread_archive_manifests - WHERE root_thread_id = ? AND status IN ('cold', 'restored') + WHERE root_thread_id = ? AND status IN ('cleanup_pending', 'cold', 'restored') ORDER BY CASE WHEN thread_id = ? THEN 1 ELSE 0 END, thread_id ASC`, [rootThreadId, rootThreadId], )) as ReadonlyArray; @@ -536,7 +605,7 @@ const make = Effect.gen(function* () { [rootThreadId, rootThreadId], )) as ReadonlyArray; for (const row of rows) { - yield* archiveImpl(ThreadId.make(String(row.thread_id))); + yield* archiveImpl(ThreadId.make(String(row.thread_id)), true); } }); @@ -631,8 +700,28 @@ const make = Effect.gen(function* () { ); }); + const getTreeSemaphore = (threadId: ThreadId) => + Effect.flatMap(resolveTreeRoot(threadId), (rootThreadId) => + SynchronizedRef.modifyEffect(threadLocksRef, (current) => { + const existing = Option.fromNullishOr(current.get(rootThreadId)); + return Option.match(existing, { + onNone: () => + Semaphore.make(1).pipe( + Effect.map((semaphore) => { + const next = new Map(current); + next.set(rootThreadId, semaphore); + return [semaphore, next] as const; + }), + ), + onSome: (semaphore) => Effect.succeed([semaphore, current] as const), + }); + }), + ); + const wrap = (operation: string, threadId: ThreadId, effect: Effect.Effect) => - effect.pipe(Effect.mapError((cause) => storageError(operation, threadId, cause))); + Effect.flatMap(getTreeSemaphore(threadId), (semaphore) => semaphore.withPermit(effect)).pipe( + Effect.mapError((cause) => storageError(operation, threadId, cause)), + ); const listIds = (query: string, operation: string) => sql.unsafe(query).pipe( @@ -643,7 +732,7 @@ const make = Effect.gen(function* () { ); return { - archiveThread: (threadId) => wrap("archive", threadId, archiveImpl(threadId)), + archiveThread: (threadId) => wrap("archive", threadId, archiveImpl(threadId, false)), restoreTree: (threadId) => wrap("restore", threadId, restoreTreeImpl(threadId)), rollbackRestoreTree: (threadId) => wrap("rollback-restore", threadId, rollbackRestoreTreeImpl(threadId)), @@ -656,7 +745,22 @@ const make = Effect.gen(function* () { Effect.mapError((cause) => storageError("compact-legacy-storage", "startup", cause)), ), listPendingArchiveThreadIds: listIds( - `SELECT thread_id FROM thread_archive_manifests WHERE status IN ('pending', 'archiving', 'cleanup_pending') ORDER BY archived_at ASC, thread_id ASC`, + `SELECT thread_id + FROM ( + SELECT thread_id, archived_at + FROM thread_archive_manifests + WHERE status IN ('pending', 'archiving', 'cleanup_pending') + UNION ALL + SELECT projection_threads.thread_id, projection_threads.archived_at + FROM projection_threads + WHERE projection_threads.archived_at IS NOT NULL + AND projection_threads.deleted_at IS NULL + AND NOT EXISTS ( + SELECT 1 FROM thread_archive_manifests + WHERE thread_archive_manifests.thread_id = projection_threads.thread_id + ) + ) + ORDER BY archived_at ASC, thread_id ASC`, "list-pending-archives", ), listPendingDeleteThreadIds: listIds( diff --git a/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts b/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts index 4023dba8d8d..7a3a907bdaf 100644 --- a/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts +++ b/apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts @@ -64,14 +64,17 @@ const make = Effect.gen(function* () { threadId, }); + const closeProviderLogWritersRequired = (threadId: ThreadDeletedEvent["payload"]["threadId"]) => + Effect.all( + [providerEventLoggers.native, providerEventLoggers.canonical].flatMap((logger) => + logger?.closeThread ? [logger.closeThread(threadId)] : [], + ), + { discard: true }, + ); + const closeProviderLogWriters = (threadId: ThreadDeletedEvent["payload"]["threadId"]) => logCleanupCauseUnlessInterrupted({ - effect: Effect.all( - [providerEventLoggers.native, providerEventLoggers.canonical].flatMap((logger) => - logger?.closeThread ? [logger.closeThread(threadId)] : [], - ), - { discard: true }, - ), + effect: closeProviderLogWritersRequired(threadId), message: "thread lifecycle cleanup skipped provider log writer close", threadId, }); @@ -84,14 +87,20 @@ const make = Effect.gen(function* () { return; } const { threadId } = job; - yield* stopProviderSession(threadId); - yield* closeThreadTerminals(threadId); - yield* closeProviderLogWriters(threadId); if (job.type === "archive") { + // Archiving must not snapshot or delete hot rows while any active writer + // can still mutate them. A failure leaves the durable archived shell or + // manifest discoverable so startup recovery can retry the boundary. + yield* providerService.stopSession({ threadId }); + yield* terminalManager.close({ threadId, deleteHistory: true }); + yield* closeProviderLogWritersRequired(threadId); yield* threadColdStorage.archiveThread(threadId); - } else { - yield* threadColdStorage.deleteThread(threadId); + return; } + yield* stopProviderSession(threadId); + yield* closeThreadTerminals(threadId); + yield* closeProviderLogWriters(threadId); + yield* threadColdStorage.deleteThread(threadId); }); const processLifecycleJobSafely = (job: ThreadLifecycleJob) => From 6ac4c9d9139d57b126d225ee7b9378e510159770 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Wed, 15 Jul 2026 21:59:45 +0100 Subject: [PATCH 05/10] Evict idle cold storage lifecycle locks - Reference-count archive-tree lock users and waiters - Remove lock entries after the final operation releases them --- .../orchestration/Layers/ThreadColdStorage.ts | 60 +++++++++++++------ 1 file changed, 43 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts index e430707010e..73803902673 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -8,7 +8,6 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; -import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; @@ -38,6 +37,14 @@ const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.UnknownFromJso const decodeUnknownJsonString = Schema.decodeUnknownEffect(Schema.UnknownFromJsonString); type SqlRow = Record; +type ThreadLockEntry = { + readonly semaphore: Semaphore.Semaphore; + readonly users: number; +}; +type AcquiredThreadLock = { + readonly rootThreadId: ThreadId; + readonly semaphore: Semaphore.Semaphore; +}; class ArchiveCodecError extends Data.TaggedError("ArchiveCodecError")<{ readonly cause: unknown; @@ -157,7 +164,7 @@ const make = Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const config = yield* ServerConfig; - const threadLocksRef = yield* SynchronizedRef.make(new Map()); + const threadLocksRef = yield* SynchronizedRef.make(new Map()); yield* sql.unsafe(`ATTACH DATABASE ? AS ${ARCHIVE_SCHEMA}`, [config.archiveDbPath]); yield* sql.unsafe(`PRAGMA ${ARCHIVE_SCHEMA}.auto_vacuum = INCREMENTAL`); @@ -703,25 +710,44 @@ const make = Effect.gen(function* () { const getTreeSemaphore = (threadId: ThreadId) => Effect.flatMap(resolveTreeRoot(threadId), (rootThreadId) => SynchronizedRef.modifyEffect(threadLocksRef, (current) => { - const existing = Option.fromNullishOr(current.get(rootThreadId)); - return Option.match(existing, { - onNone: () => - Semaphore.make(1).pipe( - Effect.map((semaphore) => { - const next = new Map(current); - next.set(rootThreadId, semaphore); - return [semaphore, next] as const; - }), - ), - onSome: (semaphore) => Effect.succeed([semaphore, current] as const), - }); + const existing = current.get(rootThreadId); + if (existing) { + const next = new Map(current); + next.set(rootThreadId, { ...existing, users: existing.users + 1 }); + return Effect.succeed([ + { rootThreadId, semaphore: existing.semaphore } satisfies AcquiredThreadLock, + next, + ] as const); + } + return Semaphore.make(1).pipe( + Effect.map((semaphore) => { + const next = new Map(current); + next.set(rootThreadId, { semaphore, users: 1 }); + return [{ rootThreadId, semaphore } satisfies AcquiredThreadLock, next] as const; + }), + ); }), ); + const releaseTreeSemaphore = (acquired: AcquiredThreadLock) => + SynchronizedRef.update(threadLocksRef, (current) => { + const existing = current.get(acquired.rootThreadId); + if (!existing || existing.semaphore !== acquired.semaphore) return current; + const next = new Map(current); + if (existing.users === 1) { + next.delete(acquired.rootThreadId); + } else { + next.set(acquired.rootThreadId, { ...existing, users: existing.users - 1 }); + } + return next; + }); + const wrap = (operation: string, threadId: ThreadId, effect: Effect.Effect) => - Effect.flatMap(getTreeSemaphore(threadId), (semaphore) => semaphore.withPermit(effect)).pipe( - Effect.mapError((cause) => storageError(operation, threadId, cause)), - ); + Effect.acquireUseRelease( + getTreeSemaphore(threadId), + ({ semaphore }) => semaphore.withPermit(effect), + releaseTreeSemaphore, + ).pipe(Effect.mapError((cause) => storageError(operation, threadId, cause))); const listIds = (query: string, operation: string) => sql.unsafe(query).pipe( From 9e5cad6cbc1068dfa6bd11db95a526afa0c451db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Wed, 15 Jul 2026 22:02:19 +0100 Subject: [PATCH 06/10] Infer cold storage service contract - Define the service members inline with Context.Service - Use the inferred Service type in the layer and orchestration test --- .../Layers/OrchestrationEngine.test.ts | 4 +- .../orchestration/Layers/ThreadColdStorage.ts | 8 +--- .../Services/ThreadColdStorage.ts | 45 ++++++++++--------- 3 files changed, 28 insertions(+), 29 deletions(-) diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 45c4bdd9dbb..d424c17859e 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -39,7 +39,7 @@ import { } from "../Services/ProjectionPipeline.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; import { ServerConfig } from "../../config.ts"; -import { ThreadColdStorage, type ThreadColdStorageShape } from "../Services/ThreadColdStorage.ts"; +import { ThreadColdStorage } from "../Services/ThreadColdStorage.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value); const asMessageId = (value: string): MessageId => MessageId.make(value); @@ -899,7 +899,7 @@ describe("OrchestrationEngine", () => { return Stream.fromIterable(events); }, }; - const coldStorage: ThreadColdStorageShape = { + const coldStorage: ThreadColdStorage["Service"] = { archiveThread: () => Effect.void, restoreTree: () => Effect.succeed(restoreOnUnarchive), rollbackRestoreTree: (threadId) => diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts index 73803902673..63d3b3a4c93 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -20,11 +20,7 @@ import { toSafeThreadAttachmentSegment, } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; -import { - ThreadColdStorage, - ThreadColdStorageError, - type ThreadColdStorageShape, -} from "../Services/ThreadColdStorage.ts"; +import { ThreadColdStorage, ThreadColdStorageError } from "../Services/ThreadColdStorage.ts"; const gzipAsync = NodeUtil.promisify(NodeZlib.gzip); const gunzipAsync = NodeUtil.promisify(NodeZlib.gunzip); @@ -793,7 +789,7 @@ const make = Effect.gen(function* () { `SELECT thread_id FROM thread_cleanup_queue WHERE reason = 'deleted' ORDER BY created_at ASC, thread_id ASC`, "list-pending-deletes", ), - } satisfies ThreadColdStorageShape; + } satisfies ThreadColdStorage["Service"]; }); export const ThreadColdStorageLive = Layer.effect(ThreadColdStorage, make); diff --git a/apps/server/src/orchestration/Services/ThreadColdStorage.ts b/apps/server/src/orchestration/Services/ThreadColdStorage.ts index acf8fe76f98..3c4b7b3f765 100644 --- a/apps/server/src/orchestration/Services/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Services/ThreadColdStorage.ts @@ -16,24 +16,27 @@ export class ThreadColdStorageError extends Schema.TaggedErrorClass Effect.Effect; - readonly restoreTree: (threadId: ThreadId) => Effect.Effect; - readonly rollbackRestoreTree: (threadId: ThreadId) => Effect.Effect; - readonly finishRestoreTree: (threadId: ThreadId) => Effect.Effect; - readonly deleteThread: (threadId: ThreadId) => Effect.Effect; - readonly removeProviderLogs: (threadId: ThreadId) => Effect.Effect; - readonly compactLegacyStorage: Effect.Effect; - readonly listPendingArchiveThreadIds: Effect.Effect< - ReadonlyArray, - ThreadColdStorageError - >; - readonly listPendingDeleteThreadIds: Effect.Effect< - ReadonlyArray, - ThreadColdStorageError - >; -} - -export class ThreadColdStorage extends Context.Service()( - "t3/orchestration/Services/ThreadColdStorage", -) {} +export class ThreadColdStorage extends Context.Service< + ThreadColdStorage, + { + readonly archiveThread: (threadId: ThreadId) => Effect.Effect; + readonly restoreTree: (threadId: ThreadId) => Effect.Effect; + readonly rollbackRestoreTree: ( + threadId: ThreadId, + ) => Effect.Effect; + readonly finishRestoreTree: (threadId: ThreadId) => Effect.Effect; + readonly deleteThread: (threadId: ThreadId) => Effect.Effect; + readonly removeProviderLogs: ( + threadId: ThreadId, + ) => Effect.Effect; + readonly compactLegacyStorage: Effect.Effect; + readonly listPendingArchiveThreadIds: Effect.Effect< + ReadonlyArray, + ThreadColdStorageError + >; + readonly listPendingDeleteThreadIds: Effect.Effect< + ReadonlyArray, + ThreadColdStorageError + >; + } +>()("t3/orchestration/Services/ThreadColdStorage") {} From a42fe41cb825f5fc516fe721d21c22e60213235c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Wed, 15 Jul 2026 23:17:19 +0100 Subject: [PATCH 07/10] Fix cold storage attachment ownership - Match archived attachments by exact persisted ids - Resume cleanup pending manifests without shell rows - Preserve attachment metadata until durable delete cleanup succeeds --- .../Layers/ThreadColdStorage.test.ts | 152 +++++++++++++++++- .../orchestration/Layers/ThreadColdStorage.ts | 56 +++++-- 2 files changed, 196 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts index 75d556e8909..7d088b679f8 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts @@ -5,6 +5,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as ServerConfig from "../../config.ts"; @@ -12,6 +13,8 @@ import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { ThreadColdStorage } from "../Services/ThreadColdStorage.ts"; import { ThreadColdStorageLive } from "./ThreadColdStorage.ts"; +const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.UnknownFromJsonString); + const insertArchivedThread = Effect.fn("insertArchivedThreadTestFixture")(function* ( threadId: ThreadId, title: string, @@ -194,6 +197,26 @@ layer("ThreadColdStorage", (it) => { const attachmentName = "thread-traversal-00000000-0000-4000-8000-000000000001.png"; yield* insertArchivedThread(threadId, "Traversal thread"); + yield* sql.unsafe( + `INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES (?, ?, NULL, 'user', 'validate attachment name', ?, 0, ?, ?)`, + [ + "message-traversal", + threadId, + encodeUnknownJsonString([ + { + type: "image", + id: attachmentName.slice(0, -4), + name: "image.png", + mimeType: "image/png", + }, + ]), + "2026-07-01T00:00:00.000Z", + "2026-07-01T00:00:00.000Z", + ], + ); yield* fs.writeFileString(path.join(config.attachmentsDir, attachmentName), "image bytes"); yield* storage.archiveThread(threadId); const escapedPath = path.join(config.attachmentsDir, "..", "thread-traversal-escape"); @@ -230,7 +253,8 @@ layer("ThreadColdStorage", (it) => { is_streaming, created_at, updated_at ) VALUES ( 'message-attachment-restore-failure', ${threadId}, NULL, 'user', 'keep me cold', - '[]', 0, '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z' + '[{"type":"image","id":"thread-attachment-restore-failure-00000000-0000-4000-8000-000000000001","name":"image.png","mimeType":"image/png"}]', + 0, '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z' ) `; yield* fs.writeFileString(path.join(config.attachmentsDir, attachmentName), "image bytes"); @@ -333,6 +357,112 @@ layer("ThreadColdStorage", (it) => { }), ); + it.effect("finishes cleanup-pending archives after their shell is removed", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const threadId = ThreadId.make("thread-cleanup-missing-shell"); + const providerLogPath = path.join(config.providerLogsDir, "thread-cleanup-missing-shell.log"); + + yield* insertArchivedThread(threadId, "Cleanup missing shell thread"); + yield* fs.makeDirectory(providerLogPath); + yield* fs.writeFileString(path.join(providerLogPath, "keep"), "force cleanup failure"); + + const archiveFailure = yield* Effect.flip(storage.archiveThread(threadId)); + assert.strictEqual(archiveFailure.operation, "archive"); + yield* sql`DELETE FROM projection_threads WHERE thread_id = ${threadId}`; + yield* fs.remove(providerLogPath, { recursive: true }); + + yield* storage.archiveThread(threadId); + const manifest = yield* sql<{ readonly status: string }>` + SELECT status FROM thread_archive_manifests WHERE thread_id = ${threadId} + `; + assert.deepStrictEqual(manifest, [{ status: "cold" }]); + assert.notDeepInclude(yield* storage.listPendingArchiveThreadIds, threadId); + }), + ); + + it.effect("archives only attachments owned by colliding thread segments", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const storage = yield* ThreadColdStorage; + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const archivedThreadId = ThreadId.make("Thread.Foo"); + const liveThreadId = ThreadId.make("thread foo"); + const archivedAttachmentId = "thread-foo-00000000-0000-4000-8000-000000000001"; + const liveAttachmentId = "thread-foo-00000000-0000-4000-8000-000000000002"; + const archivedAttachmentName = `${archivedAttachmentId}.png`; + const liveAttachmentName = `${liveAttachmentId}.png`; + + yield* insertArchivedThread(archivedThreadId, "Archived colliding thread"); + yield* insertArchivedThread(liveThreadId, "Live colliding thread"); + yield* sql` + UPDATE projection_threads SET archived_at = NULL WHERE thread_id = ${liveThreadId} + `; + yield* sql.unsafe( + `INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES (?, ?, NULL, 'user', ?, ?, 0, ?, ?)`, + [ + "message-archived-collision", + archivedThreadId, + "archive only my attachment", + encodeUnknownJsonString([ + { + type: "image", + id: archivedAttachmentId, + name: "archived.png", + mimeType: "image/png", + }, + ]), + "2026-07-01T00:00:00.000Z", + "2026-07-01T00:00:00.000Z", + ], + ); + yield* sql.unsafe( + `INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES (?, ?, NULL, 'user', ?, ?, 0, ?, ?)`, + [ + "message-live-collision", + liveThreadId, + "keep my attachment live", + encodeUnknownJsonString([ + { + type: "image", + id: liveAttachmentId, + name: "live.png", + mimeType: "image/png", + }, + ]), + "2026-07-01T00:00:00.000Z", + "2026-07-01T00:00:00.000Z", + ], + ); + const archivedAttachmentPath = path.join(config.attachmentsDir, archivedAttachmentName); + const liveAttachmentPath = path.join(config.attachmentsDir, liveAttachmentName); + yield* fs.writeFileString(archivedAttachmentPath, "archived image"); + yield* fs.writeFileString(liveAttachmentPath, "live image"); + + yield* storage.archiveThread(archivedThreadId); + + assert.isFalse(yield* fs.exists(archivedAttachmentPath)); + assert.strictEqual(yield* fs.readFileString(liveAttachmentPath), "live image"); + const archivedChunks = yield* sql<{ readonly kind: string }>` + SELECT kind FROM cold_archive.archive_thread_chunks + WHERE thread_id = ${archivedThreadId} AND kind LIKE 'attachment:%' + `; + assert.deepStrictEqual(archivedChunks, [{ kind: `attachment:${archivedAttachmentName}` }]); + }), + ); + it.effect("restores cleanup-pending bundles before unarchiving", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -462,6 +592,26 @@ layer("ThreadColdStorage", (it) => { const attachmentPath = path.join(config.attachmentsDir, attachmentName); yield* insertArchivedThread(threadId, "Delete retry thread"); + yield* sql.unsafe( + `INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, attachments_json, + is_streaming, created_at, updated_at + ) VALUES (?, ?, NULL, 'user', 'delete me', ?, 0, ?, ?)`, + [ + "message-delete-retry", + threadId, + encodeUnknownJsonString([ + { + type: "image", + id: attachmentName.slice(0, -4), + name: "delete.png", + mimeType: "image/png", + }, + ]), + "2026-07-01T00:00:00.000Z", + "2026-07-01T00:00:00.000Z", + ], + ); yield* fs.makeDirectory(attachmentPath); yield* fs.writeFileString(path.join(attachmentPath, "keep"), "force cleanup failure"); diff --git a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts index 63d3b3a4c93..bfc2b0272ed 100644 --- a/apps/server/src/orchestration/Layers/ThreadColdStorage.ts +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -16,7 +16,6 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { parseAttachmentIdFromRelativePath, - parseThreadSegmentFromAttachmentId, toSafeThreadAttachmentSegment, } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; @@ -193,8 +192,40 @@ const make = Effect.gen(function* () { const attachmentEntriesForThread = Effect.fn("attachmentEntriesForThread")(function* ( threadId: string, ) { - const segment = toSafeThreadAttachmentSegment(threadId); - if (!segment) return [] as string[]; + const attachmentIds = new Set(); + const attachmentRows = (yield* sql.unsafe( + `SELECT attachments_json + FROM projection_thread_messages + WHERE thread_id = ? AND attachments_json IS NOT NULL`, + [threadId], + )) as ReadonlyArray; + for (const row of attachmentRows) { + const attachments = yield* decodeUnknownJsonString(String(row.attachments_json)); + if (!Array.isArray(attachments)) continue; + for (const attachment of attachments) { + if (attachment === null || typeof attachment !== "object" || Array.isArray(attachment)) { + continue; + } + const id = (attachment as Record).id; + if (typeof id === "string" && id.length > 0) { + attachmentIds.add(id); + } + } + } + + const archivedEntries = new Set( + ( + (yield* sql.unsafe( + `SELECT kind + FROM ${ARCHIVE_SCHEMA}.archive_thread_chunks + WHERE thread_id = ? AND kind LIKE 'attachment:%'`, + [threadId], + )) as ReadonlyArray + ).flatMap((row) => { + const entry = String(row.kind).slice("attachment:".length); + return isSafeAttachmentEntry(entry) ? [entry] : []; + }), + ); const entries = yield* fs .readDirectory(config.attachmentsDir, { recursive: false }) .pipe( @@ -203,8 +234,9 @@ const make = Effect.gen(function* () { ), ); return entries.filter((entry) => { + if (archivedEntries.has(entry)) return true; const attachmentId = parseAttachmentIdFromRelativePath(entry); - return attachmentId !== null && parseThreadSegmentFromAttachmentId(attachmentId) === segment; + return attachmentId !== null && attachmentIds.has(attachmentId); }); }); @@ -317,17 +349,17 @@ const make = Effect.gen(function* () { const manifest = manifestRows[0]; const source = manifest ?? threadRows[0]; if (!source) return; + if (source.status === "cold") return; + if (source.status === "cleanup_pending") { + yield* completeArchiveCleanup(threadId); + return; + } if (threadRows.length === 0) { if (source.status === "pending" || source.status === "archiving") { yield* discardIncompleteArchive(threadId); } return; } - if (source.status === "cold") return; - if (source.status === "cleanup_pending") { - yield* completeArchiveCleanup(threadId); - return; - } if (source.status === "restored" && !allowRestored) return; const rootThreadId = String(source.root_thread_id ?? threadId); const archivedAt = String(source.archived_at ?? DateTime.formatIso(yield* DateTime.now)); @@ -646,6 +678,10 @@ const make = Effect.gen(function* () { VALUES (?, 'deleted', CURRENT_TIMESTAMP)`, [threadId], ); + // Keep the hot rows or cold chunks available until external cleanup has + // succeeded so an interrupted delete can recover exact attachment owners. + yield* removeAttachments(threadId); + yield* removeProviderLogsImpl(threadId); yield* sql.withTransaction( Effect.gen(function* () { yield* sql.unsafe( @@ -674,8 +710,6 @@ const make = Effect.gen(function* () { yield* sql.unsafe(`DELETE FROM thread_archive_manifests WHERE thread_id = ?`, [threadId]); }), ); - yield* removeAttachments(threadId); - yield* removeProviderLogsImpl(threadId); yield* reclaimFreePages(); yield* sql.unsafe(`DELETE FROM thread_cleanup_queue WHERE thread_id = ?`, [threadId]); }); From cebc5c4775d49b99eb4a4481f01792bef4d157c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Thu, 16 Jul 2026 21:14:22 +0100 Subject: [PATCH 08/10] Preserve fork storage migration metadata --- apps/mobile/app.config.ts | 6 +++--- apps/server/src/persistence/Migrations.ts | 8 ++++---- ...cle.test.ts => 035_036_ThreadStorageLifecycle.test.ts} | 4 ++-- ...{033_ThreadColdArchive.ts => 035_ThreadColdArchive.ts} | 0 ...adCleanupQueue.ts => 036_DeletedThreadCleanupQueue.ts} | 0 5 files changed, 9 insertions(+), 9 deletions(-) rename apps/server/src/persistence/Migrations/{033_034_ThreadStorageLifecycle.test.ts => 035_036_ThreadStorageLifecycle.test.ts} (95%) rename apps/server/src/persistence/Migrations/{033_ThreadColdArchive.ts => 035_ThreadColdArchive.ts} (100%) rename apps/server/src/persistence/Migrations/{034_DeletedThreadCleanupQueue.ts => 036_DeletedThreadCleanupQueue.ts} (100%) diff --git a/apps/mobile/app.config.ts b/apps/mobile/app.config.ts index c27be2e357f..e06c92dbcf0 100644 --- a/apps/mobile/app.config.ts +++ b/apps/mobile/app.config.ts @@ -160,7 +160,7 @@ const config: ExpoConfig = { userInterfaceStyle: "automatic", updates: { enabled: true, - url: "https://u.expo.dev/d763fcb8-d37c-41ea-a773-b54a0ab4a454", + url: "https://u.expo.dev/c65ac46d-6488-49af-b61e-ab9bef78f96e", checkAutomatically: "ON_LOAD", fallbackToCacheTimeout: 0, }, @@ -331,10 +331,10 @@ const config: ExpoConfig = { tracesToken: repoEnv.EXPO_PUBLIC_OTLP_TRACES_TOKEN ?? null, }, eas: { - projectId: "d763fcb8-d37c-41ea-a773-b54a0ab4a454", + projectId: "c65ac46d-6488-49af-b61e-ab9bef78f96e", }, }, - owner: "pingdotgg", + owner: "quicksaver", }; export default config; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 244aa9d00dd..8870f16ca6c 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -45,8 +45,8 @@ import Migration0029 from "./Migrations/029_ProjectionThreadDetailOrderingIndexe import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes.ts"; import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; -import Migration0033 from "./Migrations/033_ThreadColdArchive.ts"; -import Migration0034 from "./Migrations/034_DeletedThreadCleanupQueue.ts"; +import Migration0035 from "./Migrations/035_ThreadColdArchive.ts"; +import Migration0036 from "./Migrations/036_DeletedThreadCleanupQueue.ts"; /** * Migration loader with all migrations defined inline. @@ -91,8 +91,8 @@ export const migrationEntries = [ [30, "ProjectionThreadShellArchiveIndexes", Migration0030], [31, "AuthAuthorizationScopes", Migration0031], [32, "AuthPairingProofKeyThumbprint", Migration0032], - [33, "ThreadColdArchive", Migration0033], - [34, "DeletedThreadCleanupQueue", Migration0034], + [35, "ThreadColdArchive", Migration0035], + [36, "DeletedThreadCleanupQueue", Migration0036], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/033_034_ThreadStorageLifecycle.test.ts b/apps/server/src/persistence/Migrations/035_036_ThreadStorageLifecycle.test.ts similarity index 95% rename from apps/server/src/persistence/Migrations/033_034_ThreadStorageLifecycle.test.ts rename to apps/server/src/persistence/Migrations/035_036_ThreadStorageLifecycle.test.ts index 17619226a70..7b8cff58d54 100644 --- a/apps/server/src/persistence/Migrations/033_034_ThreadStorageLifecycle.test.ts +++ b/apps/server/src/persistence/Migrations/035_036_ThreadStorageLifecycle.test.ts @@ -8,7 +8,7 @@ import * as NodeSqliteClient from "../NodeSqliteClient.ts"; const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); -layer("033-034 thread storage lifecycle migrations", (it) => { +layer("035-036 thread storage lifecycle migrations", (it) => { it.effect("queues archived and deleted threads", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -41,7 +41,7 @@ layer("033-034 thread storage lifecycle migrations", (it) => { deletedAt: "2026-07-03T00:00:00.000Z", }); - yield* runMigrations({ toMigrationInclusive: 34 }); + yield* runMigrations({ toMigrationInclusive: 36 }); const manifests = yield* sql<{ readonly threadId: string; diff --git a/apps/server/src/persistence/Migrations/033_ThreadColdArchive.ts b/apps/server/src/persistence/Migrations/035_ThreadColdArchive.ts similarity index 100% rename from apps/server/src/persistence/Migrations/033_ThreadColdArchive.ts rename to apps/server/src/persistence/Migrations/035_ThreadColdArchive.ts diff --git a/apps/server/src/persistence/Migrations/034_DeletedThreadCleanupQueue.ts b/apps/server/src/persistence/Migrations/036_DeletedThreadCleanupQueue.ts similarity index 100% rename from apps/server/src/persistence/Migrations/034_DeletedThreadCleanupQueue.ts rename to apps/server/src/persistence/Migrations/036_DeletedThreadCleanupQueue.ts From 70cc4f11c4bc0cbdaa0face2dc2daebaf969cad4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Thu, 16 Jul 2026 22:45:38 +0100 Subject: [PATCH 09/10] Remove unrelated fork deployment metadata --- apps/mobile/app.config.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/mobile/app.config.ts b/apps/mobile/app.config.ts index e06c92dbcf0..c27be2e357f 100644 --- a/apps/mobile/app.config.ts +++ b/apps/mobile/app.config.ts @@ -160,7 +160,7 @@ const config: ExpoConfig = { userInterfaceStyle: "automatic", updates: { enabled: true, - url: "https://u.expo.dev/c65ac46d-6488-49af-b61e-ab9bef78f96e", + url: "https://u.expo.dev/d763fcb8-d37c-41ea-a773-b54a0ab4a454", checkAutomatically: "ON_LOAD", fallbackToCacheTimeout: 0, }, @@ -331,10 +331,10 @@ const config: ExpoConfig = { tracesToken: repoEnv.EXPO_PUBLIC_OTLP_TRACES_TOKEN ?? null, }, eas: { - projectId: "c65ac46d-6488-49af-b61e-ab9bef78f96e", + projectId: "d763fcb8-d37c-41ea-a773-b54a0ab4a454", }, }, - owner: "quicksaver", + owner: "pingdotgg", }; export default config; From 1ace10b62c42ba169720304263b485b2b409b96e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Fri, 17 Jul 2026 00:27:34 +0100 Subject: [PATCH 10/10] Hide optimistically archived sidebar threads - Reuse archive filtering for project rows and navigation - Cover persisted and optimistic archive visibility --- apps/web/src/components/Sidebar.logic.test.ts | 23 +++++++++++++++++++ apps/web/src/components/Sidebar.logic.ts | 13 +++++++++++ apps/web/src/components/Sidebar.tsx | 23 +++++++++---------- 3 files changed, 47 insertions(+), 12 deletions(-) diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index 574e33d4dab..4facf5eeed7 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -1,6 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; import { createThreadJumpHintVisibilityController, + filterVisibleSidebarThreads, getSidebarThreadIdsToPrewarm, getVisibleSidebarThreadIds, resolveAdjacentThreadId, @@ -28,6 +29,7 @@ import { ProviderInstanceId, ThreadId, } from "@t3tools/contracts"; +import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; import { DEFAULT_INTERACTION_MODE, DEFAULT_RUNTIME_MODE, @@ -841,6 +843,27 @@ function makeThread(overrides: Partial = {}): Thread { }; } +describe("filterVisibleSidebarThreads", () => { + it("excludes archived shells and optimistically archived threads", () => { + const visibleThread = makeThread({ id: ThreadId.make("thread-visible") }); + const optimisticThread = makeThread({ id: ThreadId.make("thread-optimistic") }); + const archivedThread = makeThread({ + id: ThreadId.make("thread-archived"), + archivedAt: "2026-03-09T10:05:00.000Z", + }); + const optimisticThreadKey = scopedThreadKey( + scopeThreadRef(optimisticThread.environmentId, optimisticThread.id), + ); + + expect( + filterVisibleSidebarThreads( + [visibleThread, optimisticThread, archivedThread], + new Set([optimisticThreadKey]), + ).map((thread) => thread.id), + ).toEqual([visibleThread.id]); + }); +}); + describe("getFallbackThreadIdAfterDelete", () => { it("returns the top remaining thread in the deleted thread's project sidebar order", () => { const fallbackThreadId = getFallbackThreadIdAfterDelete({ diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index 4e7614ed551..50ef8aea6f3 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -1,5 +1,6 @@ import * as React from "react"; import type { SidebarProjectSortOrder, SidebarThreadSortOrder } from "@t3tools/contracts/settings"; +import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; import { getThreadSortTimestamp, sortThreads, @@ -490,6 +491,18 @@ export function getVisibleThreadsForProject>(input: }; } +export function filterVisibleSidebarThreads< + T extends Pick, +>(threads: readonly T[], optimisticallyArchivedThreadKeys: ReadonlySet): T[] { + return threads.filter( + (thread) => + thread.archivedAt === null && + !optimisticallyArchivedThreadKeys.has( + scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), + ), + ); +} + export function getFallbackThreadIdAfterDelete< T extends Pick & ThreadSortInput, >(input: { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index ec25726a8bd..ba39773b33e 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -185,6 +185,7 @@ import { import { useThreadSelectionStore } from "../threadSelectionStore"; import { useOpenAddProjectCommandPalette } from "../commandPaletteContext"; import { + filterVisibleSidebarThreads, getSidebarThreadIdsToPrewarm, resolveAdjacentThreadId, isContextMenuPointerDown, @@ -1177,6 +1178,9 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec }); const openPrLink = useOpenPrLink(); const sidebarThreads = useThreadShellsForProjectRefs(project.memberProjectRefs); + const optimisticallyArchivedThreadKeys = useOptimisticThreadArchiveStore( + (state) => state.threadKeys, + ); const sidebarThreadByKey = useMemo( () => new Map( @@ -1267,7 +1271,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec }); }; const visibleProjectThreads = sortThreads( - projectThreads.filter((thread) => thread.archivedAt === null), + filterVisibleSidebarThreads(projectThreads, optimisticallyArchivedThreadKeys), threadSortOrder, ); const projectStatus = resolveProjectStatusIndicator( @@ -1280,7 +1284,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec projectStatus, visibleProjectThreads, }; - }, [projectThreads, threadLastVisitedAts, threadSortOrder]); + }, [optimisticallyArchivedThreadKeys, projectThreads, threadLastVisitedAts, threadSortOrder]); const pinnedCollapsedThread = useMemo(() => { const activeThreadKey = activeRouteThreadKey ?? undefined; if (!activeThreadKey || projectExpanded) { @@ -3373,14 +3377,7 @@ export default function Sidebar() { }, []); const visibleThreads = useMemo( - () => - sidebarThreads.filter( - (thread) => - thread.archivedAt === null && - !optimisticallyArchivedThreadKeys.has( - scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), - ), - ), + () => filterVisibleSidebarThreads(sidebarThreads, optimisticallyArchivedThreadKeys), [optimisticallyArchivedThreadKeys, sidebarThreads], ); const sortedProjects = useMemo(() => { @@ -3419,8 +3416,9 @@ export default function Sidebar() { () => sortedProjects.flatMap((project) => { const projectThreads = sortThreads( - (threadsByProjectKey.get(project.projectKey) ?? []).filter( - (thread) => thread.archivedAt === null, + filterVisibleSidebarThreads( + threadsByProjectKey.get(project.projectKey) ?? [], + optimisticallyArchivedThreadKeys, ), sidebarThreadSortOrder, ); @@ -3456,6 +3454,7 @@ export default function Sidebar() { sidebarThreadSortOrder, sidebarThreadPreviewCount, expandedThreadListsByProject, + optimisticallyArchivedThreadKeys, projectExpandedById, routeThreadKey, sortedProjects,