From 63859aa072249c64153ef20faeb46018154724b8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 8 May 2026 15:49:41 -0700 Subject: [PATCH] perf: Add archived shell snapshot support (#2593) Co-authored-by: Cursor Agent --- .../Layers/CheckpointDiffQuery.test.ts | 6 + .../Layers/OrchestrationEngine.test.ts | 7 + .../Layers/ProjectionSnapshotQuery.test.ts | 120 ++++++++ .../Layers/ProjectionSnapshotQuery.ts | 278 +++++++++++++++++- .../Services/ProjectionSnapshotQuery.ts | 11 + apps/server/src/persistence/Migrations.ts | 2 + ...030_ProjectionThreadShellArchiveIndexes.ts | 16 + .../Layers/ProjectSetupScriptRunner.test.ts | 1 + .../Layers/ProviderSessionReaper.test.ts | 1 + apps/server/src/server.test.ts | 7 + apps/server/src/serverRuntimeStartup.test.ts | 3 + apps/server/src/ws.ts | 29 ++ apps/web/src/components/ChatView.browser.tsx | 3 + .../settings/ConnectionsSettings.tsx | 28 +- .../settings/ProviderInstanceCard.tsx | 2 +- .../components/settings/SettingsPanels.tsx | 150 ++++++---- .../settings/SourceControlSettings.tsx | 2 +- .../components/settings/settingsLayout.tsx | 6 +- apps/web/src/environmentApi.ts | 1 + apps/web/src/hooks/useThreadActions.ts | 34 ++- apps/web/src/lib/archivedThreadsState.ts | 105 +++++++ apps/web/src/rpc/wsRpcClient.ts | 7 + packages/contracts/src/ipc.ts | 2 + packages/contracts/src/orchestration.ts | 5 + packages/contracts/src/rpc.ts | 10 + 25 files changed, 761 insertions(+), 75 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/030_ProjectionThreadShellArchiveIndexes.ts create mode 100644 apps/web/src/lib/archivedThreadsState.ts diff --git a/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts index 7e012312dcabeae9dcf80d8ebcf88910c1f1984d..5d4b072c94dd2ce5b032a3682a5dcf3d77be2133 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts @@ -94,6 +94,8 @@ describe("CheckpointDiffQueryLive", () => { Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), + getArchivedShellSnapshot: () => + Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }), getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), @@ -174,6 +176,8 @@ describe("CheckpointDiffQueryLive", () => { Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), + getArchivedShellSnapshot: () => + Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }), getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), @@ -222,6 +226,8 @@ describe("CheckpointDiffQueryLive", () => { Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"), + getArchivedShellSnapshot: () => + Effect.die("CheckpointDiffQuery should not request archived shell snapshots"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }), getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index d74c2c5ed421c244debb4ada26df6712f696c5fe..75fd236f9ef40140638ed49e48c2aa7172012710 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -183,6 +183,13 @@ describe("OrchestrationEngine", () => { threads: [], updatedAt: projectionSnapshot.updatedAt, }), + getArchivedShellSnapshot: () => + Effect.succeed({ + snapshotSequence: projectionSnapshot.snapshotSequence, + projects: [], + threads: [], + updatedAt: projectionSnapshot.updatedAt, + }), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: projectionSnapshot.snapshotSequence }), getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }), diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 7f6f973ed8969503685e933c4d0dee27810fb1fb..01d7c2957468e94c14e3016d2baa2148390bd6c8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -441,6 +441,126 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }), ); + it.effect("keeps archived threads out of the main shell snapshot", () => + Effect.gen(function* () { + const snapshotQuery = yield* ProjectionSnapshotQuery; + const sql = yield* SqlClient.SqlClient; + + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_state`; + + yield* sql` + INSERT INTO projection_projects ( + project_id, + title, + workspace_root, + default_model_selection_json, + scripts_json, + created_at, + updated_at, + deleted_at + ) + VALUES ( + 'project-archive-test', + 'Archive Test', + '/tmp/archive-test', + '{"provider":"codex","model":"gpt-5-codex"}', + '[]', + '2026-04-06T00:00:00.000Z', + '2026-04-06T00:00:01.000Z', + NULL + ) + `; + + yield* sql` + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + latest_turn_id, + latest_user_message_at, + pending_approval_count, + pending_user_input_count, + has_actionable_proposed_plan, + created_at, + updated_at, + archived_at, + deleted_at + ) + VALUES + ( + 'thread-active', + 'project-archive-test', + 'Active Thread', + '{"provider":"codex","model":"gpt-5-codex"}', + 'full-access', + 'default', + NULL, + NULL, + NULL, + NULL, + 0, + 0, + 0, + '2026-04-06T00:00:02.000Z', + '2026-04-06T00:00:03.000Z', + NULL, + NULL + ), + ( + 'thread-archived', + 'project-archive-test', + 'Archived Thread', + '{"provider":"codex","model":"gpt-5-codex"}', + 'full-access', + 'default', + NULL, + NULL, + NULL, + NULL, + 0, + 0, + 0, + '2026-04-06T00:00:04.000Z', + '2026-04-06T00:00:05.000Z', + '2026-04-06T00:00:06.000Z', + NULL + ) + `; + + yield* sql` + INSERT INTO projection_state (projector, last_applied_sequence, updated_at) + VALUES + (${ORCHESTRATION_PROJECTOR_NAMES.projects}, 4, '2026-04-06T00:00:07.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threads}, 4, '2026-04-06T00:00:07.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadMessages}, 4, '2026-04-06T00:00:07.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans}, 4, '2026-04-06T00:00:07.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadActivities}, 4, '2026-04-06T00:00:07.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadSessions}, 4, '2026-04-06T00:00:07.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.checkpoints}, 4, '2026-04-06T00:00:07.000Z') + `; + + const shellSnapshot = yield* snapshotQuery.getShellSnapshot(); + assert.deepEqual( + shellSnapshot.threads.map((thread) => thread.id), + [ThreadId.make("thread-active")], + ); + + const archivedShellSnapshot = yield* snapshotQuery.getArchivedShellSnapshot(); + assert.deepEqual( + archivedShellSnapshot.threads.map((thread) => thread.id), + [ThreadId.make("thread-archived")], + ); + assert.equal(archivedShellSnapshot.threads[0]?.archivedAt, "2026-04-06T00:00:06.000Z"); + }), + ); + it.effect( "reads targeted project, thread, and count queries without hydrating the full snapshot", () => diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 9a539d77aec4a5a851343512ee518cd6a9b501e8..cf6cb7beaa8458f1783301e5afeadb80b325cef5 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -327,6 +327,66 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const listActiveThreadRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionThreadDbRowSchema, + execute: () => + sql` + SELECT + thread_id AS "threadId", + project_id AS "projectId", + title, + model_selection_json AS "modelSelection", + runtime_mode AS "runtimeMode", + interaction_mode AS "interactionMode", + branch, + worktree_path AS "worktreePath", + latest_turn_id AS "latestTurnId", + created_at AS "createdAt", + updated_at AS "updatedAt", + archived_at AS "archivedAt", + latest_user_message_at AS "latestUserMessageAt", + pending_approval_count AS "pendingApprovalCount", + pending_user_input_count AS "pendingUserInputCount", + has_actionable_proposed_plan AS "hasActionableProposedPlan", + deleted_at AS "deletedAt" + FROM projection_threads + WHERE deleted_at IS NULL + AND archived_at IS NULL + ORDER BY project_id ASC, created_at ASC, thread_id ASC + `, + }); + + const listArchivedThreadRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionThreadDbRowSchema, + execute: () => + sql` + SELECT + thread_id AS "threadId", + project_id AS "projectId", + title, + model_selection_json AS "modelSelection", + runtime_mode AS "runtimeMode", + interaction_mode AS "interactionMode", + branch, + worktree_path AS "worktreePath", + latest_turn_id AS "latestTurnId", + created_at AS "createdAt", + updated_at AS "updatedAt", + archived_at AS "archivedAt", + latest_user_message_at AS "latestUserMessageAt", + pending_approval_count AS "pendingApprovalCount", + pending_user_input_count AS "pendingUserInputCount", + has_actionable_proposed_plan AS "hasActionableProposedPlan", + deleted_at AS "deletedAt" + FROM projection_threads + WHERE deleted_at IS NULL + AND archived_at IS NOT NULL + ORDER BY project_id ASC, archived_at DESC, thread_id DESC + `, + }); + const listThreadMessageRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadMessageDbRowSchema, @@ -411,6 +471,56 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const listActiveThreadSessionRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionThreadSessionDbRowSchema, + execute: () => + sql` + SELECT + sessions.thread_id AS "threadId", + sessions.status, + sessions.provider_name AS "providerName", + sessions.provider_instance_id AS "providerInstanceId", + sessions.provider_session_id AS "providerSessionId", + sessions.provider_thread_id AS "providerThreadId", + sessions.runtime_mode AS "runtimeMode", + sessions.active_turn_id AS "activeTurnId", + sessions.last_error AS "lastError", + sessions.updated_at AS "updatedAt" + FROM projection_thread_sessions sessions + INNER JOIN projection_threads threads + ON threads.thread_id = sessions.thread_id + WHERE threads.deleted_at IS NULL + AND threads.archived_at IS NULL + ORDER BY sessions.thread_id ASC + `, + }); + + const listArchivedThreadSessionRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionThreadSessionDbRowSchema, + execute: () => + sql` + SELECT + sessions.thread_id AS "threadId", + sessions.status, + sessions.provider_name AS "providerName", + sessions.provider_instance_id AS "providerInstanceId", + sessions.provider_session_id AS "providerSessionId", + sessions.provider_thread_id AS "providerThreadId", + sessions.runtime_mode AS "runtimeMode", + sessions.active_turn_id AS "activeTurnId", + sessions.last_error AS "lastError", + sessions.updated_at AS "updatedAt" + FROM projection_thread_sessions sessions + INNER JOIN projection_threads threads + ON threads.thread_id = sessions.thread_id + WHERE threads.deleted_at IS NULL + AND threads.archived_at IS NOT NULL + ORDER BY sessions.thread_id ASC + `, + }); + const listCheckpointRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionCheckpointDbRowSchema, @@ -475,6 +585,33 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ON turns.thread_id = threads.thread_id AND turns.turn_id = threads.latest_turn_id WHERE threads.deleted_at IS NULL + AND threads.archived_at IS NULL + AND threads.latest_turn_id IS NOT NULL + ORDER BY turns.thread_id ASC + `, + }); + + const listArchivedLatestTurnRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionLatestTurnDbRowSchema, + execute: () => + sql` + SELECT + turns.thread_id AS "threadId", + turns.turn_id AS "turnId", + turns.state, + turns.requested_at AS "requestedAt", + turns.started_at AS "startedAt", + turns.completed_at AS "completedAt", + turns.assistant_message_id AS "assistantMessageId", + turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", + turns.source_proposed_plan_id AS "sourceProposedPlanId" + FROM projection_threads threads + JOIN projection_turns turns + ON turns.thread_id = threads.thread_id + AND turns.turn_id = threads.latest_turn_id + WHERE threads.deleted_at IS NULL + AND threads.archived_at IS NOT NULL AND threads.latest_turn_id IS NOT NULL ORDER BY turns.thread_id ASC `, @@ -557,6 +694,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { FROM projection_threads WHERE project_id = ${projectId} AND deleted_at IS NULL + AND archived_at IS NULL ORDER BY created_at ASC, thread_id ASC LIMIT 1 `, @@ -607,6 +745,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { FROM projection_threads WHERE thread_id = ${threadId} AND deleted_at IS NULL + AND archived_at IS NULL LIMIT 1 `, }); @@ -717,6 +856,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { AND turns.turn_id = threads.latest_turn_id WHERE threads.thread_id = ${threadId} AND threads.deleted_at IS NULL + AND threads.archived_at IS NULL LIMIT 1 `, }); @@ -1232,7 +1372,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), - listThreadRows(undefined).pipe( + listActiveThreadRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getShellSnapshot:listThreads:query", @@ -1240,7 +1380,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), - listThreadSessionRows(undefined).pipe( + listActiveThreadSessionRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getShellSnapshot:listThreadSessions:query", @@ -1350,6 +1490,139 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { }), ); + const getArchivedShellSnapshot: ProjectionSnapshotQueryShape["getArchivedShellSnapshot"] = () => + sql + .withTransaction( + Effect.all([ + listProjectRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjects:query", + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjects:decodeRows", + ), + ), + ), + listArchivedThreadRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreads:query", + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreads:decodeRows", + ), + ), + ), + listArchivedThreadSessionRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreadSessions:query", + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreadSessions:decodeRows", + ), + ), + ), + listArchivedLatestTurnRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listLatestTurns:query", + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listLatestTurns:decodeRows", + ), + ), + ), + listProjectionStateRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjectionState:query", + "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjectionState:decodeRows", + ), + ), + ), + ]), + ) + .pipe( + Effect.flatMap(([projectRows, threadRows, sessionRows, latestTurnRows, stateRows]) => + Effect.gen(function* () { + let updatedAt: string | null = null; + for (const row of projectRows) { + updatedAt = maxIso(updatedAt, row.updatedAt); + } + for (const row of threadRows) { + updatedAt = maxIso(updatedAt, row.updatedAt); + } + for (const row of sessionRows) { + updatedAt = maxIso(updatedAt, row.updatedAt); + } + for (const row of latestTurnRows) { + updatedAt = maxIso(updatedAt, row.requestedAt); + if (row.startedAt !== null) { + updatedAt = maxIso(updatedAt, row.startedAt); + } + if (row.completedAt !== null) { + updatedAt = maxIso(updatedAt, row.completedAt); + } + } + for (const row of stateRows) { + updatedAt = maxIso(updatedAt, row.updatedAt); + } + + const activeProjectIds = new Set(threadRows.map((row) => row.projectId)); + const repositoryIdentities = yield* resolveRepositoryIdentitiesForProjects( + projectRows.filter((row) => activeProjectIds.has(row.projectId)), + ); + const latestTurnByThread = new Map( + latestTurnRows.map((row) => [row.threadId, mapLatestTurn(row)] as const), + ); + const sessionByThread = new Map( + sessionRows.map((row) => [row.threadId, mapSessionRow(row)] as const), + ); + + const snapshot = { + snapshotSequence: computeSnapshotSequence(stateRows), + projects: projectRows + .filter((row) => row.deletedAt === null && activeProjectIds.has(row.projectId)) + .map((row) => + mapProjectShellRow(row, repositoryIdentities.get(row.projectId) ?? null), + ), + threads: threadRows.map( + (row): OrchestrationThreadShell => ({ + id: row.threadId, + projectId: row.projectId, + title: row.title, + modelSelection: row.modelSelection, + runtimeMode: row.runtimeMode, + interactionMode: row.interactionMode, + branch: row.branch, + worktreePath: row.worktreePath, + latestTurn: latestTurnByThread.get(row.threadId) ?? null, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + archivedAt: row.archivedAt, + session: sessionByThread.get(row.threadId) ?? null, + latestUserMessageAt: row.latestUserMessageAt, + hasPendingApprovals: row.pendingApprovalCount > 0, + hasPendingUserInput: row.pendingUserInputCount > 0, + hasActionableProposedPlan: row.hasActionableProposedPlan > 0, + }), + ), + updatedAt: updatedAt ?? new Date(0).toISOString(), + }; + + return yield* decodeShellSnapshot(snapshot).pipe( + Effect.mapError( + toPersistenceDecodeError( + "ProjectionSnapshotQuery.getArchivedShellSnapshot:decodeShellSnapshot", + ), + ), + ); + }), + ), + Effect.mapError((error) => { + if (isPersistenceError(error)) { + return error; + } + return toPersistenceSqlError("ProjectionSnapshotQuery.getArchivedShellSnapshot:query")( + error, + ); + }), + ); + const getSnapshotSequence: ProjectionSnapshotQueryShape["getSnapshotSequence"] = () => listProjectionStateRows(undefined).pipe( Effect.mapError( @@ -1683,6 +1956,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { getCommandReadModel, getSnapshot, getShellSnapshot, + getArchivedShellSnapshot, getSnapshotSequence, getCounts, getActiveProjectByWorkspaceRoot, diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 2df890d547ede536e52891ff74013312e4c46fc9..53dfe33bf492d0cc23b95e059aaf7ee354b6ee98 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -72,6 +72,17 @@ export interface ProjectionSnapshotQueryShape { ProjectionRepositoryError >; + /** + * Read archived thread shell summaries for the archive page. + * + * This query is separate from the main shell snapshot so archived threads + * are never bootstrapped into normal navigation state. + */ + readonly getArchivedShellSnapshot: () => Effect.Effect< + OrchestrationShellSnapshot, + ProjectionRepositoryError + >; + /** * Read the latest projection snapshot sequence without hydrating read-model * entities. diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index c0918de8493bacb2447d3e81dc5c637e566b894a..cc5024d5f51b50d1f2352abc6ed5ddbb88033562 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -42,6 +42,7 @@ import Migration0026 from "./Migrations/026_CanonicalizeModelSelectionOptions.ts import Migration0027 from "./Migrations/027_ProviderSessionRuntimeInstanceId.ts"; import Migration0028 from "./Migrations/028_ProjectionThreadSessionInstanceId.ts"; import Migration0029 from "./Migrations/029_ProjectionThreadDetailOrderingIndexes.ts"; +import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes.ts"; /** * Migration loader with all migrations defined inline. @@ -83,6 +84,7 @@ export const migrationEntries = [ [27, "ProviderSessionRuntimeInstanceId", Migration0027], [28, "ProjectionThreadSessionInstanceId", Migration0028], [29, "ProjectionThreadDetailOrderingIndexes", Migration0029], + [30, "ProjectionThreadShellArchiveIndexes", Migration0030], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/030_ProjectionThreadShellArchiveIndexes.ts b/apps/server/src/persistence/Migrations/030_ProjectionThreadShellArchiveIndexes.ts new file mode 100644 index 0000000000000000000000000000000000000000..3b7bf51f04b0e0a712266d363d0eacd0c51a8ec5 --- /dev/null +++ b/apps/server/src/persistence/Migrations/030_ProjectionThreadShellArchiveIndexes.ts @@ -0,0 +1,16 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_threads_shell_active + ON projection_threads(deleted_at, archived_at, project_id, created_at, thread_id) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_threads_shell_archived + ON projection_threads(deleted_at, archived_at, project_id, thread_id) + `; +}); diff --git a/apps/server/src/project/Layers/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/Layers/ProjectSetupScriptRunner.test.ts index 72e3b5e6d5d952cdff64b67db0711ad0f30c6384..1a13229fd8e37fdd040aba3b163bd4a16e772ac5 100644 --- a/apps/server/src/project/Layers/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/Layers/ProjectSetupScriptRunner.test.ts @@ -25,6 +25,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), + getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), getCounts: () => Effect.die("unused"), getActiveProjectByWorkspaceRoot: (workspaceRoot) => diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 57903e3776cd0c1e280afd52890d9944801f4221..add225e2eea7639c640e573f71f82669b2fe39b5 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -194,6 +194,7 @@ describe("ProviderSessionReaper", () => { getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), + getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: input.readModel.snapshotSequence }), getCounts: () => Effect.die("unused"), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 819abb979da39c14300d37b8d68220531b82c5f0..857da7aebe45d606c4199a421263db0b9940bd17 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -630,6 +630,13 @@ const buildAppUnderTest = (options?: { threads: [], updatedAt: "1970-01-01T00:00:00.000Z", }), + getArchivedShellSnapshot: () => + Effect.succeed({ + snapshotSequence: 0, + projects: [], + threads: [], + updatedAt: new Date(0).toISOString(), + }), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), getProjectShellById: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index b47c7d0af6a7edb1fd5074c46818f71d337d496a..1da9a63070fdb3014d766f627d0cbba06f7204d1 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -85,6 +85,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), + getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Deferred.await(releaseCounts).pipe( @@ -139,6 +140,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), + getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), getActiveProjectByWorkspaceRoot: () => @@ -191,6 +193,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), + getArchivedShellSnapshot: () => Effect.die("unused"), getSnapshotSequence: () => Effect.die("unused"), getCounts: () => Effect.die("unused"), getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 80f5b1f8e3c33a82e24650b347f14aba82c56c97..d3c891b2bb05cc7c6ab2c9cdea652178444aece2 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -308,6 +308,7 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => }), ); case "thread.deleted": + case "thread.archived": return Effect.succeed( Option.some({ kind: "thread-removed" as const, @@ -315,6 +316,17 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => threadId: event.payload.threadId, }), ); + case "thread.unarchived": + return projectionSnapshotQuery.getThreadShellById(event.payload.threadId).pipe( + Effect.map((thread) => + Option.map(thread, (nextThread) => ({ + kind: "thread-upserted" as const, + sequence: event.sequence, + thread: nextThread, + })), + ), + Effect.catch(() => Effect.succeed(Option.none())), + ); default: if (event.aggregateKind !== "thread") { return Effect.succeed(Option.none()); @@ -734,6 +746,23 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => }), { "rpc.aggregate": "orchestration" }, ), + [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]: (_input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, + projectionSnapshotQuery.getArchivedShellSnapshot().pipe( + Effect.tapError((cause) => + Effect.logError("orchestration archived shell snapshot load failed", { cause }), + ), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load archived orchestration shell snapshot", + cause, + }), + ), + ), + { "rpc.aggregate": "orchestration" }, + ), [ORCHESTRATION_WS_METHODS.subscribeThread]: (input) => observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeThread, diff --git a/apps/web/src/components/ChatView.browser.tsx b/apps/web/src/components/ChatView.browser.tsx index 17cf014c225bdf9bee8799ef388e91349a81a258..f51780abf5852f5359b08a0147e76feb3e9936f5 100644 --- a/apps/web/src/components/ChatView.browser.tsx +++ b/apps/web/src/components/ChatView.browser.tsx @@ -218,6 +218,9 @@ function createMockEnvironmentApi(input: { getFullThreadDiff: (() => { throw new Error("Not implemented in browser test."); }) as EnvironmentApi["orchestration"]["getFullThreadDiff"], + getArchivedShellSnapshot: (() => { + throw new Error("Not implemented in browser test."); + }) as EnvironmentApi["orchestration"]["getArchivedShellSnapshot"], subscribeShell: (() => () => undefined) as EnvironmentApi["orchestration"]["subscribeShell"], subscribeThread: (() => () => undefined) as EnvironmentApi["orchestration"]["subscribeThread"], diff --git a/apps/web/src/components/settings/ConnectionsSettings.tsx b/apps/web/src/components/settings/ConnectionsSettings.tsx index b3dea6c31156647421f1e2104600c6689c707011..17dc177a981e7cf444c3581e93f5ea221d87bc08 100644 --- a/apps/web/src/components/settings/ConnectionsSettings.tsx +++ b/apps/web/src/components/settings/ConnectionsSettings.tsx @@ -2662,14 +2662,26 @@ export function ConnectionsSettings() { } }} > - - - Add environment - - } - /> + + + + Add environment + + } + /> + } + /> + Add environment + Add Environment diff --git a/apps/web/src/components/settings/ProviderInstanceCard.tsx b/apps/web/src/components/settings/ProviderInstanceCard.tsx index 44ba093615f0f24da7266b469a1a0355a86563a1..9a00ccb75aee38632461b8b7cee92f77f5a31fbe 100644 --- a/apps/web/src/components/settings/ProviderInstanceCard.tsx +++ b/apps/web/src/components/settings/ProviderInstanceCard.tsx @@ -648,7 +648,7 @@ export function ProviderInstanceCard({ ); const authRowNode = ( -

+

{hasAuthenticatedEmail ? ( <> Authenticated as diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index 9f0166a9d2a261ad94b005836e3799d57e333bab..def96b9ba9dd090f816ac328486220ee29f6af59 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -44,14 +44,10 @@ import { } from "../../providerInstances"; import { ensureLocalApi, readLocalApi } from "../../localApi"; import { useShallow } from "zustand/react/shallow"; -import { - selectProjectsAcrossEnvironments, - selectThreadShellsAcrossEnvironments, - useStore, -} from "../../store"; +import { selectProjectsAcrossEnvironments, useStore } from "../../store"; +import { useArchivedThreadSnapshots } from "../../lib/archivedThreadsState"; import { formatRelativeTime, formatRelativeTimeLabel } from "../../timestampFormat"; import { Button } from "../ui/button"; -import { Empty, EmptyDescription, EmptyHeader, EmptyMedia, EmptyTitle } from "../ui/empty"; import { DraftInput } from "../ui/draft-input"; import { Select, SelectItem, SelectPopup, SelectTrigger, SelectValue } from "../ui/select"; import { Switch } from "../ui/switch"; @@ -1330,14 +1326,50 @@ export function ProviderSettingsPanel() { export function ArchivedThreadsPanel() { const projects = useStore(useShallow(selectProjectsAcrossEnvironments)); - const threads = useStore(useShallow(selectThreadShellsAcrossEnvironments)); const { unarchiveThread, confirmAndDeleteThread } = useThreadActions(); + const environmentIds = useMemo( + () => [...new Set(projects.map((project) => project.environmentId))], + [projects], + ); + const { + snapshots: archivedSnapshots, + error: archiveError, + isLoading: isLoadingArchive, + refresh: refreshArchivedThreads, + } = useArchivedThreadSnapshots(environmentIds); + const archivedGroups = useMemo(() => { - return projects + const projectsByEnvironmentAndId = new Map( + archivedSnapshots.flatMap(({ environmentId, snapshot }) => + snapshot.projects.map( + (project) => + [ + `${environmentId}:${project.id}`, + { + id: project.id, + environmentId, + name: project.title, + cwd: project.workspaceRoot, + }, + ] as const, + ), + ), + ); + const threads = archivedSnapshots.flatMap(({ environmentId, snapshot }) => + snapshot.threads.map((thread) => ({ + ...thread, + environmentId, + })), + ); + + return [...projectsByEnvironmentAndId.values()] .map((project) => ({ project, threads: threads - .filter((thread) => thread.projectId === project.id && thread.archivedAt !== null) + .filter( + (thread) => + thread.projectId === project.id && thread.environmentId === project.environmentId, + ) .toSorted((left, right) => { const leftKey = left.archivedAt ?? left.createdAt; const rightKey = right.archivedAt ?? right.createdAt; @@ -1345,7 +1377,7 @@ export function ArchivedThreadsPanel() { }), })) .filter((group) => group.threads.length > 0); - }, [projects, threads]); + }, [archivedSnapshots]); const handleArchivedThreadContextMenu = useCallback( async (threadRef: ScopedThreadRef, position: { x: number; y: number }) => { @@ -1362,6 +1394,7 @@ export function ArchivedThreadsPanel() { if (clicked === "unarchive") { try { await unarchiveThread(threadRef); + refreshArchivedThreads(); } catch (error) { toastManager.add( stackedThreadToast({ @@ -1376,24 +1409,37 @@ export function ArchivedThreadsPanel() { if (clicked === "delete") { await confirmAndDeleteThread(threadRef); + refreshArchivedThreads(); } }, - [confirmAndDeleteThread, unarchiveThread], + [confirmAndDeleteThread, refreshArchivedThreads, unarchiveThread], ); return ( {archivedGroups.length === 0 ? ( - - - - - - No archived threads - Archived threads will appear here. - - + + {isLoadingArchive ? ( + + ) : ( + + )} + {isLoadingArchive + ? "Loading archived threads" + : archiveError + ? "Could not load archived threads" + : "No archived threads"} + + } + description={ + isLoadingArchive + ? "Checking connected environments." + : (archiveError ?? "Archived threads will appear here.") + } + /> ) : ( archivedGroups.map(({ project, threads: projectThreads }) => ( @@ -1403,9 +1449,8 @@ export function ArchivedThreadsPanel() { icon={} > {projectThreads.map((thread) => ( -

{ event.preventDefault(); void handleArchivedThreadContextMenu( @@ -1416,39 +1461,40 @@ export function ArchivedThreadsPanel() { }, ); }} - > -
-

{thread.title}

-

+ title={thread.title} + description={ + <> Archived {formatRelativeTimeLabel(thread.archivedAt ?? thread.createdAt)} {" \u00b7 Created "} {formatRelativeTimeLabel(thread.createdAt)} -

-
- -
+ + } + control={ + + } + /> ))} )) diff --git a/apps/web/src/components/settings/SourceControlSettings.tsx b/apps/web/src/components/settings/SourceControlSettings.tsx index 0518848bc583a11908d2d7336189b0324b446281..59c6e4aa63da349df6ed18c53ad06b880c70da75 100644 --- a/apps/web/src/components/settings/SourceControlSettings.tsx +++ b/apps/web/src/components/settings/SourceControlSettings.tsx @@ -226,7 +226,7 @@ function DiscoveryItemRow({ ) : null} -

+

{itemSummary({ item, auth, authAccount })}

diff --git a/apps/web/src/components/settings/settingsLayout.tsx b/apps/web/src/components/settings/settingsLayout.tsx index 54c2c99b8657cb1d7a82e04d8a2df7dceab8c82c..391898b1110bff5a7271b9f5697621dc86b36e43 100644 --- a/apps/web/src/components/settings/settingsLayout.tsx +++ b/apps/web/src/components/settings/settingsLayout.tsx @@ -52,7 +52,9 @@ export function SettingsRow({ resetAction, control, children, -}: { + className, + ...rowProps +}: Omit, "title"> & { title: ReactNode; description: ReactNode; status?: ReactNode; @@ -62,9 +64,11 @@ export function SettingsRow({ }) { return (
diff --git a/apps/web/src/environmentApi.ts b/apps/web/src/environmentApi.ts index 335d68d3720ab4fff85c75e8b23f25ee0bdc74c1..9e7b0c939a9eb892cd46c7bd7ef9a534f57fb66f 100644 --- a/apps/web/src/environmentApi.ts +++ b/apps/web/src/environmentApi.ts @@ -47,6 +47,7 @@ export function createEnvironmentApi(rpcClient: WsRpcClient): EnvironmentApi { dispatchCommand: rpcClient.orchestration.dispatchCommand, getTurnDiff: rpcClient.orchestration.getTurnDiff, getFullThreadDiff: rpcClient.orchestration.getFullThreadDiff, + getArchivedShellSnapshot: rpcClient.orchestration.getArchivedShellSnapshot, subscribeShell: (callback, options) => rpcClient.orchestration.subscribeShell(callback, options), subscribeThread: (input, callback, options) => diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 46a1955cc49df7ed2ac6a0bc808e5da3a4b183ce..7fc4bb2efd16848c17fa68a287b9949a722661f9 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -9,6 +9,7 @@ import { useComposerDraftStore } from "../composerDraftStore"; import { useNewThreadHandler } from "./useHandleNewThread"; import { ensureEnvironmentApi, readEnvironmentApi } from "../environmentApi"; import { invalidateGitQueries } from "../lib/gitReactQuery"; +import { refreshArchivedThreadsForEnvironment } from "../lib/archivedThreadsState"; import { newCommandId } from "../lib/utils"; import { readLocalApi } from "../localApi"; import { @@ -68,19 +69,22 @@ export function useThreadActions() { throw new Error("Cannot archive a running thread."); } - await api.orchestration.dispatchCommand({ + const currentRouteThreadRef = getCurrentRouteThreadRef(); + const shouldNavigateToDraft = + currentRouteThreadRef?.threadId === threadRef.threadId && + currentRouteThreadRef.environmentId === threadRef.environmentId; + const archiveCommand = api.orchestration.dispatchCommand({ type: "thread.archive", commandId: newCommandId(), threadId: threadRef.threadId, }); - const currentRouteThreadRef = getCurrentRouteThreadRef(); - if ( - currentRouteThreadRef?.threadId === threadRef.threadId && - currentRouteThreadRef.environmentId === threadRef.environmentId - ) { + if (shouldNavigateToDraft) { await handleNewThreadRef.current(scopeProjectRef(thread.environmentId, thread.projectId)); } + + await archiveCommand; + refreshArchivedThreadsForEnvironment(threadRef.environmentId); }, [getCurrentRouteThreadRef, resolveThreadTarget], ); @@ -93,6 +97,7 @@ export function useThreadActions() { commandId: newCommandId(), threadId: target.threadId, }); + refreshArchivedThreadsForEnvironment(target.environmentId); }, []); const deleteThread = useCallback( @@ -100,7 +105,16 @@ export function useThreadActions() { const api = readEnvironmentApi(target.environmentId); if (!api) return; const resolved = resolveThreadTarget(target); - if (!resolved) return; + if (!resolved) { + // Thread not in main store (e.g. archived thread) — dispatch delete directly. + await api.orchestration.dispatchCommand({ + type: "thread.delete", + commandId: newCommandId(), + threadId: target.threadId, + }); + refreshArchivedThreadsForEnvironment(target.environmentId); + return; + } const { thread, threadRef } = resolved; const state = useStore.getState(); const threads = selectThreadsForEnvironment(state, threadRef.environmentId); @@ -175,6 +189,7 @@ export function useThreadActions() { commandId: newCommandId(), threadId: threadRef.threadId, }); + refreshArchivedThreadsForEnvironment(threadRef.environmentId); clearComposerDraftForThread(threadRef); clearProjectDraftThreadById( scopeProjectRef(threadRef.environmentId, thread.projectId), @@ -252,13 +267,12 @@ export function useThreadActions() { if (!api) return; const localApi = readLocalApi(); const resolved = resolveThreadTarget(target); - if (!resolved) return; - const { thread } = resolved; if (confirmThreadDelete && localApi) { + const title = resolved?.thread.title ?? "this thread"; const confirmed = await localApi.dialogs.confirm( [ - `Delete thread "${thread.title}"?`, + `Delete thread "${title}"?`, "This permanently clears conversation history for this thread.", ].join("\n"), ); diff --git a/apps/web/src/lib/archivedThreadsState.ts b/apps/web/src/lib/archivedThreadsState.ts new file mode 100644 index 0000000000000000000000000000000000000000..c978e7690eb29d62a3f5e6266f3f90539e28d7fe --- /dev/null +++ b/apps/web/src/lib/archivedThreadsState.ts @@ -0,0 +1,105 @@ +import { useAtomValue } from "@effect/atom-react"; +import { EnvironmentId, type OrchestrationShellSnapshot } from "@t3tools/contracts"; +import { Cause, Effect, Option } from "effect"; +import { AsyncResult, Atom } from "effect/unstable/reactivity"; +import { useCallback, useMemo } from "react"; + +import { readEnvironmentApi } from "../environmentApi"; +import { appAtomRegistry } from "../rpc/atomRegistry"; + +const ARCHIVED_THREADS_STALE_TIME_MS = 5_000; +const ARCHIVED_THREADS_IDLE_TTL_MS = 5 * 60_000; +const ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR = "\u001f"; + +export type ArchivedSnapshotEntry = { + readonly environmentId: EnvironmentId; + readonly snapshot: OrchestrationShellSnapshot; +}; + +const knownArchivedThreadEnvironmentKeys = new Set(); + +function makeArchivedThreadsEnvironmentKey(environmentIds: ReadonlyArray): string { + return environmentIds.toSorted().join(ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR); +} + +function parseArchivedThreadsEnvironmentKey(key: string): ReadonlyArray { + if (key.length === 0) { + return []; + } + return key + .split(ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR) + .map((environmentId) => EnvironmentId.make(environmentId)); +} + +const archivedThreadSnapshotsAtom = Atom.family((environmentKey: string) => { + knownArchivedThreadEnvironmentKeys.add(environmentKey); + return Atom.make( + Effect.promise(async (): Promise> => { + const environmentIds = parseArchivedThreadsEnvironmentKey(environmentKey); + const snapshots = await Promise.all( + environmentIds.map(async (environmentId) => { + const api = readEnvironmentApi(environmentId); + if (!api) { + return null; + } + return { + environmentId, + snapshot: await api.orchestration.getArchivedShellSnapshot(), + }; + }), + ); + return snapshots.filter((snapshot) => snapshot !== null); + }), + ).pipe( + Atom.swr({ + staleTime: ARCHIVED_THREADS_STALE_TIME_MS, + revalidateOnMount: true, + }), + Atom.setIdleTTL(ARCHIVED_THREADS_IDLE_TTL_MS), + Atom.withLabel(`archived-thread-snapshots:${environmentKey}`), + ); +}); + +function readArchivedThreadsError( + result: AsyncResult.AsyncResult, unknown>, +): string | null { + if (result._tag !== "Failure") { + return null; + } + + const error = Cause.squash(result.cause); + return error instanceof Error ? error.message : "Failed to load archived threads."; +} + +export function refreshArchivedThreadsForEnvironment(environmentId: EnvironmentId): void { + for (const key of knownArchivedThreadEnvironmentKeys) { + if (parseArchivedThreadsEnvironmentKey(key).includes(environmentId)) { + appAtomRegistry.refresh(archivedThreadSnapshotsAtom(key)); + } + } +} + +export function useArchivedThreadSnapshots(environmentIds: ReadonlyArray): { + readonly snapshots: ReadonlyArray; + readonly error: string | null; + readonly isLoading: boolean; + readonly refresh: () => void; +} { + const environmentKey = useMemo( + () => makeArchivedThreadsEnvironmentKey(environmentIds), + [environmentIds], + ); + const atom = archivedThreadSnapshotsAtom(environmentKey); + const result = useAtomValue(atom); + const snapshots = Option.getOrElse(AsyncResult.value(result), () => []); + const refresh = useCallback(() => { + appAtomRegistry.refresh(atom); + }, [atom]); + + return { + snapshots, + error: readArchivedThreadsError(result), + isLoading: result.waiting, + refresh, + }; +} diff --git a/apps/web/src/rpc/wsRpcClient.ts b/apps/web/src/rpc/wsRpcClient.ts index b36ffb0ab2e20d9f6ece28cfd7a5a11706ca59c6..2f1ca624d986af65b874a56e514c17b68c74c15f 100644 --- a/apps/web/src/rpc/wsRpcClient.ts +++ b/apps/web/src/rpc/wsRpcClient.ts @@ -143,6 +143,9 @@ export interface WsRpcClient { readonly dispatchCommand: RpcUnaryMethod; readonly getTurnDiff: RpcUnaryMethod; readonly getFullThreadDiff: RpcUnaryMethod; + readonly getArchivedShellSnapshot: RpcUnaryNoArgMethod< + typeof ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot + >; readonly subscribeShell: RpcStreamMethod; readonly subscribeThread: RpcInputStreamMethod; }; @@ -288,6 +291,10 @@ export function createWsRpcClient(transport: WsTransport): WsRpcClient { transport.request((client) => client[ORCHESTRATION_WS_METHODS.getTurnDiff](input)), getFullThreadDiff: (input) => transport.request((client) => client[ORCHESTRATION_WS_METHODS.getFullThreadDiff](input)), + getArchivedShellSnapshot: () => + transport.request((client) => + client[ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]({}), + ), subscribeShell: (listener, options) => transport.subscribe( (client) => client[ORCHESTRATION_WS_METHODS.subscribeShell]({}), diff --git a/packages/contracts/src/ipc.ts b/packages/contracts/src/ipc.ts index abe8af5022f8171dec25a6e744a6a14020130dc9..58894adac1adcfce9bff5618e17f312641979813 100644 --- a/packages/contracts/src/ipc.ts +++ b/packages/contracts/src/ipc.ts @@ -55,6 +55,7 @@ import type { OrchestrationGetFullThreadDiffResult, OrchestrationGetTurnDiffInput, OrchestrationGetTurnDiffResult, + OrchestrationShellSnapshot, OrchestrationShellStreamItem, OrchestrationSubscribeThreadInput, OrchestrationThreadStreamItem, @@ -544,6 +545,7 @@ export interface EnvironmentApi { getFullThreadDiff: ( input: OrchestrationGetFullThreadDiffInput, ) => Promise; + getArchivedShellSnapshot: () => Promise; subscribeShell: ( callback: (event: OrchestrationShellStreamItem) => void, options?: { diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 44d840d1499f38b36a6921f58ef8043e63cc6889..401928171c8bb90a399e3ade7243e415466e60f8 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -27,6 +27,7 @@ export const ORCHESTRATION_WS_METHODS = { getTurnDiff: "orchestration.getTurnDiff", getFullThreadDiff: "orchestration.getFullThreadDiff", replayEvents: "orchestration.replayEvents", + getArchivedShellSnapshot: "orchestration.getArchivedShellSnapshot", subscribeShell: "orchestration.subscribeShell", subscribeThread: "orchestration.subscribeThread", } as const; @@ -1222,6 +1223,10 @@ export const OrchestrationRpcSchemas = { input: OrchestrationReplayEventsInput, output: OrchestrationReplayEventsResult, }, + getArchivedShellSnapshot: { + input: Schema.Struct({}), + output: OrchestrationShellSnapshot, + }, subscribeThread: { input: OrchestrationSubscribeThreadInput, output: OrchestrationThreadStreamItem, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index eb72b14f9e53c3d310d8cce572d1caddb9f3cc6e..705621b5dac34734b4c6383bca6a1bd1ddb0d37d 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -410,6 +410,15 @@ export const WsOrchestrationReplayEventsRpc = Rpc.make(ORCHESTRATION_WS_METHODS. error: OrchestrationReplayEventsError, }); +export const WsOrchestrationGetArchivedShellSnapshotRpc = Rpc.make( + ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, + { + payload: OrchestrationRpcSchemas.getArchivedShellSnapshot.input, + success: OrchestrationRpcSchemas.getArchivedShellSnapshot.output, + error: OrchestrationGetSnapshotError, + }, +); + export const WsOrchestrationSubscribeShellRpc = Rpc.make(ORCHESTRATION_WS_METHODS.subscribeShell, { payload: OrchestrationRpcSchemas.subscribeShell.input, success: OrchestrationRpcSchemas.subscribeShell.output, @@ -497,6 +506,7 @@ export const WsRpcGroup = RpcGroup.make( WsOrchestrationGetTurnDiffRpc, WsOrchestrationGetFullThreadDiffRpc, WsOrchestrationReplayEventsRpc, + WsOrchestrationGetArchivedShellSnapshotRpc, WsOrchestrationSubscribeShellRpc, WsOrchestrationSubscribeThreadRpc, ); -- 2.54.0