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..7f455002c96 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; @@ -400,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 ( undefined : props.onUnarchive, }} simultaneousWithExternalGesture={props.simultaneousSwipeGesture} threadTitle={props.thread.title} @@ -434,7 +438,16 @@ function ArchivedThreadRow(props: { }} > - + {props.isUnarchiving ? ( + + ) : ( + + )} @@ -500,6 +513,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 +578,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 +596,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.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 405103450e8..d424c17859e 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -9,6 +9,7 @@ import { type OrchestrationEvent, ProviderInstanceId, } from "@t3tools/contracts"; +import { it as effectIt } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -38,6 +39,7 @@ import { } from "../Services/ProjectionPipeline.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; import { ServerConfig } from "../../config.ts"; +import { ThreadColdStorage } from "../Services/ThreadColdStorage.ts"; const asProjectId = (value: string): ProjectId => 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: ThreadColdStorage["Service"] = { + 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 9598d1c3ef5..6e1defce881 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); @@ -105,6 +107,8 @@ const makeOrchestrationEngine = Effect.gen(function* () { const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => { 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, @@ -150,6 +154,25 @@ 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) { + restoredUnarchiveThreadId = unarchiveThreadId; + commandReadModel = yield* projectionSnapshotQuery.getCommandReadModel(); + } + } + const eventBase = yield* decideOrchestrationCommand({ command: envelope.command, readModel: commandReadModel, @@ -212,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) { @@ -261,6 +295,40 @@ const makeOrchestrationEngine = Effect.gen(function* () { return; } + if ( + restoredUnarchiveThreadId !== null && + !restoredUnarchiveCommitted && + 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 new file mode 100644 index 00000000000..7d088b679f8 --- /dev/null +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.test.ts @@ -0,0 +1,633 @@ +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 Schema from "effect/Schema"; +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 encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.UnknownFromJsonString); + +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 + ) 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' + ) + `; +}); + +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("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; + 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)); + + // 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} + `; + 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} + `; + 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" }]); + }), + ); + + 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* 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"); + yield* sql` + UPDATE cold_archive.archive_thread_chunks + SET kind = 'attachment:../thread-traversal-escape' + 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" }]); + 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', + '[{"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"); + 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); + }), + ); + + 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("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; + 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; + 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* 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"); + + 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 new file mode 100644 index 00000000000..bfc2b0272ed --- /dev/null +++ b/apps/server/src/orchestration/Layers/ThreadColdStorage.ts @@ -0,0 +1,829 @@ +// @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 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 { + parseAttachmentIdFromRelativePath, + toSafeThreadAttachmentSegment, +} from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; +import { ThreadColdStorage, ThreadColdStorageError } 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; +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; +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; +}> {} + +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): 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 }), + ), + ); +} + +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) => + 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; + 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`); + 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 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( + Effect.catch((error) => + error.reason._tag === "NotFound" ? Effect.succeed([] as string[]) : Effect.fail(error), + ), + ); + return entries.filter((entry) => { + if (archivedEntries.has(entry)) return true; + const attachmentId = parseAttachmentIdFromRelativePath(entry); + return attachmentId !== null && attachmentIds.has(attachmentId); + }); + }); + + 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.catch((error) => + error.reason._tag === "NotFound" ? Effect.succeed([] as string[]) : Effect.fail(error), + ), + ); + yield* Effect.forEach( + 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 }, + ); + }); + + const reclaimFreePages = Effect.fn("reclaimThreadStorageFreePages")(function* () { + yield* sql.unsafe("PRAGMA main.incremental_vacuum(2048)"); + 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; + 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 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 + 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 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 === "restored" && !allowRestored) 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 = yield* 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; + } + + // Chunk creation stays retryable outside the hot-row deletion transaction. + // A retry replaces every partial chunk before deleting source data. + 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, + 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 = 'cleanup_pending', original_bytes = ?, compressed_bytes = ?, + updated_at = CURRENT_TIMESTAMP, error = NULL + WHERE thread_id = ?`, + [originalBytes, compressedBytes, threadId], + ); + return true; + }), + ); + + if (!archivedAtDestructiveBoundary) return; + + yield* completeArchiveCleanup(threadId); + }); + + const insertRows = Effect.fn("restoreArchiveRows")(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) { + // 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 ${quoteIdentifier(table)} (${columns.map(quoteIdentifier).join(", ")}) VALUES (${placeholders})`, + columns.map((column) => row[column]), + ); + } + }); + + const restoreThread = Effect.fn("restoreArchivedThread")(function* (threadId: ThreadId) { + const bundleRows = (yield* sql.unsafe( + `SELECT 1 AS present FROM ${ARCHIVE_SCHEMA}.archive_threads WHERE thread_id = ? LIMIT 1`, + [threadId], + )) as ReadonlyArray; + 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* () { + 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 + SET status = 'restored', updated_at = CURRENT_TIMESTAMP, error = NULL + WHERE thread_id = ?`, + [threadId], + ); + }), + ); + + 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 ('cleanup_pending', '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 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)), true); + } + }); + + 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], + ); + // 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( + `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* reclaimFreePages(); + yield* sql.unsafe(`DELETE FROM thread_cleanup_queue WHERE thread_id = ?`, [threadId]); + }); + + 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 getTreeSemaphore = (threadId: ThreadId) => + Effect.flatMap(resolveTreeRoot(threadId), (rootThreadId) => + SynchronizedRef.modifyEffect(threadLocksRef, (current) => { + 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.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( + 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, false)), + 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)), + 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 ( + 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( + `SELECT thread_id FROM thread_cleanup_queue WHERE reason = 'deleted' ORDER BY created_at ASC, thread_id ASC`, + "list-pending-deletes", + ), + } satisfies ThreadColdStorage["Service"]; +}); + +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..7a3a907bdaf 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,97 @@ const make = Effect.gen(function* () { threadId, }); - const processThreadDeleted = Effect.fn("processThreadDeleted")(function* ( - event: ThreadDeletedEvent, + 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: closeProviderLogWritersRequired(threadId), + 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; + 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); + return; + } yield* stopProviderSession(threadId); yield* closeThreadTerminals(threadId); + yield* closeProviderLogWriters(threadId); + 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 }); } - return worker.enqueue(event); + if (event.type === "thread.archived") { + return worker.enqueue({ type: "archive", threadId: event.payload.threadId }); + } + 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..3c4b7b3f765 --- /dev/null +++ b/apps/server/src/orchestration/Services/ThreadColdStorage.ts @@ -0,0 +1,42 @@ +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 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") {} 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..8870f16ca6c 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 Migration0035 from "./Migrations/035_ThreadColdArchive.ts"; +import Migration0036 from "./Migrations/036_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], + [35, "ThreadColdArchive", Migration0035], + [36, "DeletedThreadCleanupQueue", Migration0036], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/035_036_ThreadStorageLifecycle.test.ts b/apps/server/src/persistence/Migrations/035_036_ThreadStorageLifecycle.test.ts new file mode 100644 index 00000000000..7b8cff58d54 --- /dev/null +++ b/apps/server/src/persistence/Migrations/035_036_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("035-036 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: 36 }); + + 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/035_ThreadColdArchive.ts b/apps/server/src/persistence/Migrations/035_ThreadColdArchive.ts new file mode 100644 index 00000000000..d10e7e5394b --- /dev/null +++ b/apps/server/src/persistence/Migrations/035_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/036_DeletedThreadCleanupQueue.ts b/apps/server/src/persistence/Migrations/036_DeletedThreadCleanupQueue.ts new file mode 100644 index 00000000000..24cc80ec44e --- /dev/null +++ b/apps/server/src/persistence/Migrations/036_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.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 6a76494daa3..ba39773b33e 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, @@ -184,6 +185,7 @@ import { import { useThreadSelectionStore } from "../threadSelectionStore"; import { useOpenAddProjectCommandPalette } from "../commandPaletteContext"; import { + filterVisibleSidebarThreads, getSidebarThreadIdsToPrewarm, resolveAdjacentThreadId, isContextMenuPointerDown, @@ -1176,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( @@ -1266,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( @@ -1279,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) { @@ -3107,6 +3112,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); @@ -3369,8 +3377,8 @@ export default function Sidebar() { }, []); const visibleThreads = useMemo( - () => sidebarThreads.filter((thread) => thread.archivedAt === null), - [sidebarThreads], + () => filterVisibleSidebarThreads(sidebarThreads, optimisticallyArchivedThreadKeys), + [optimisticallyArchivedThreadKeys, sidebarThreads], ); const sortedProjects = useMemo(() => { const sortableProjects = sidebarProjects.map((project) => ({ @@ -3408,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, ); @@ -3445,6 +3454,7 @@ export default function Sidebar() { sidebarThreadSortOrder, sidebarThreadPreviewCount, expandedThreadListsByProject, + optimisticallyArchivedThreadKeys, projectExpandedById, routeThreadKey, sortedProjects, diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index 40017d56314..95ec76b694a 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,77 +1585,68 @@ export function ArchivedThreadsPanel() { title={project.name} icon={} > - {projectThreads.map((thread) => ( - { - event.preventDefault(); - 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.", - }), - ); - } - })(); - }} - title={thread.title} - description={ - <> - Archived {formatRelativeTimeLabel(thread.archivedAt ?? thread.createdAt)} - {" \u00b7 Created "} - {formatRelativeTimeLabel(thread.createdAt)} - - } - control={ - - } - /> - ))} + } + })(); + }} + title={thread.title} + description={ + <> + Archived {formatRelativeTimeLabel(thread.archivedAt ?? thread.createdAt)} + {" \u00b7 Created "} + {formatRelativeTimeLabel(thread.createdAt)} + + } + control={ + + } + /> + ); + })} )) )} diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 07655ad30d7..0eca6109d06 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,36 @@ 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); + void archivePromise.then((archiveResult) => { + if (archiveResult._tag === "Success") { + refreshArchivedThreadsForEnvironment(threadRef.environmentId); + } + }); 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 +150,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); +}