authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-05-08 15:49:41-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-05-08 15:49:41-07:00
log63859aa072249c64153ef20faeb46018154724b8
treeb0f1b726f5b9c6cdf2719f0083338a3d39aaa750
parent7bfacd55d89524ee2bce68e5ae863222c4609f68
signature Signed by PGP key B5690EEEBB952194

perf: Add archived shell snapshot support (#2593)

Co-authored-by: Cursor Agent <cursoragent@cursor.com>

25 files changed, 761 insertions(+), 75 deletions(-)

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts+6
......@@ -94,6 +94,8 @@ describe("CheckpointDiffQueryLive", () => {
9494 Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
9595 getShellSnapshot: () =>
9696 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
97 getArchivedShellSnapshot: () =>
98 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
9799 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
98100 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
99101 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
......@@ -174,6 +176,8 @@ describe("CheckpointDiffQueryLive", () => {
174176 Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
175177 getShellSnapshot: () =>
176178 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
179 getArchivedShellSnapshot: () =>
180 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
177181 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
178182 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
179183 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
......@@ -222,6 +226,8 @@ describe("CheckpointDiffQueryLive", () => {
222226 Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
223227 getShellSnapshot: () =>
224228 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
229 getArchivedShellSnapshot: () =>
230 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
225231 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
226232 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
227233 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+7
......@@ -183,6 +183,13 @@ describe("OrchestrationEngine", () => {
183183 threads: [],
184184 updatedAt: projectionSnapshot.updatedAt,
185185 }),
186 getArchivedShellSnapshot: () =>
187 Effect.succeed({
188 snapshotSequence: projectionSnapshot.snapshotSequence,
189 projects: [],
190 threads: [],
191 updatedAt: projectionSnapshot.updatedAt,
192 }),
186193 getSnapshotSequence: () =>
187194 Effect.succeed({ snapshotSequence: projectionSnapshot.snapshotSequence }),
188195 getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+120
......@@ -441,6 +441,126 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
441441 }),
442442 );
443443
444 it.effect("keeps archived threads out of the main shell snapshot", () =>
445 Effect.gen(function* () {
446 const snapshotQuery = yield* ProjectionSnapshotQuery;
447 const sql = yield* SqlClient.SqlClient;
448
449 yield* sql`DELETE FROM projection_projects`;
450 yield* sql`DELETE FROM projection_threads`;
451 yield* sql`DELETE FROM projection_state`;
452
453 yield* sql`
454 INSERT INTO projection_projects (
455 project_id,
456 title,
457 workspace_root,
458 default_model_selection_json,
459 scripts_json,
460 created_at,
461 updated_at,
462 deleted_at
463 )
464 VALUES (
465 'project-archive-test',
466 'Archive Test',
467 '/tmp/archive-test',
468 '{"provider":"codex","model":"gpt-5-codex"}',
469 '[]',
470 '2026-04-06T00:00:00.000Z',
471 '2026-04-06T00:00:01.000Z',
472 NULL
473 )
474 `;
475
476 yield* sql`
477 INSERT INTO projection_threads (
478 thread_id,
479 project_id,
480 title,
481 model_selection_json,
482 runtime_mode,
483 interaction_mode,
484 branch,
485 worktree_path,
486 latest_turn_id,
487 latest_user_message_at,
488 pending_approval_count,
489 pending_user_input_count,
490 has_actionable_proposed_plan,
491 created_at,
492 updated_at,
493 archived_at,
494 deleted_at
495 )
496 VALUES
497 (
498 'thread-active',
499 'project-archive-test',
500 'Active Thread',
501 '{"provider":"codex","model":"gpt-5-codex"}',
502 'full-access',
503 'default',
504 NULL,
505 NULL,
506 NULL,
507 NULL,
508 0,
509 0,
510 0,
511 '2026-04-06T00:00:02.000Z',
512 '2026-04-06T00:00:03.000Z',
513 NULL,
514 NULL
515 ),
516 (
517 'thread-archived',
518 'project-archive-test',
519 'Archived Thread',
520 '{"provider":"codex","model":"gpt-5-codex"}',
521 'full-access',
522 'default',
523 NULL,
524 NULL,
525 NULL,
526 NULL,
527 0,
528 0,
529 0,
530 '2026-04-06T00:00:04.000Z',
531 '2026-04-06T00:00:05.000Z',
532 '2026-04-06T00:00:06.000Z',
533 NULL
534 )
535 `;
536
537 yield* sql`
538 INSERT INTO projection_state (projector, last_applied_sequence, updated_at)
539 VALUES
540 (${ORCHESTRATION_PROJECTOR_NAMES.projects}, 4, '2026-04-06T00:00:07.000Z'),
541 (${ORCHESTRATION_PROJECTOR_NAMES.threads}, 4, '2026-04-06T00:00:07.000Z'),
542 (${ORCHESTRATION_PROJECTOR_NAMES.threadMessages}, 4, '2026-04-06T00:00:07.000Z'),
543 (${ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans}, 4, '2026-04-06T00:00:07.000Z'),
544 (${ORCHESTRATION_PROJECTOR_NAMES.threadActivities}, 4, '2026-04-06T00:00:07.000Z'),
545 (${ORCHESTRATION_PROJECTOR_NAMES.threadSessions}, 4, '2026-04-06T00:00:07.000Z'),
546 (${ORCHESTRATION_PROJECTOR_NAMES.checkpoints}, 4, '2026-04-06T00:00:07.000Z')
547 `;
548
549 const shellSnapshot = yield* snapshotQuery.getShellSnapshot();
550 assert.deepEqual(
551 shellSnapshot.threads.map((thread) => thread.id),
552 [ThreadId.make("thread-active")],
553 );
554
555 const archivedShellSnapshot = yield* snapshotQuery.getArchivedShellSnapshot();
556 assert.deepEqual(
557 archivedShellSnapshot.threads.map((thread) => thread.id),
558 [ThreadId.make("thread-archived")],
559 );
560 assert.equal(archivedShellSnapshot.threads[0]?.archivedAt, "2026-04-06T00:00:06.000Z");
561 }),
562 );
563
444564 it.effect(
445565 "reads targeted project, thread, and count queries without hydrating the full snapshot",
446566 () =>
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+276-2
......@@ -327,6 +327,66 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
327327 `,
328328 });
329329
330 const listActiveThreadRows = SqlSchema.findAll({
331 Request: Schema.Void,
332 Result: ProjectionThreadDbRowSchema,
333 execute: () =>
334 sql`
335 SELECT
336 thread_id AS "threadId",
337 project_id AS "projectId",
338 title,
339 model_selection_json AS "modelSelection",
340 runtime_mode AS "runtimeMode",
341 interaction_mode AS "interactionMode",
342 branch,
343 worktree_path AS "worktreePath",
344 latest_turn_id AS "latestTurnId",
345 created_at AS "createdAt",
346 updated_at AS "updatedAt",
347 archived_at AS "archivedAt",
348 latest_user_message_at AS "latestUserMessageAt",
349 pending_approval_count AS "pendingApprovalCount",
350 pending_user_input_count AS "pendingUserInputCount",
351 has_actionable_proposed_plan AS "hasActionableProposedPlan",
352 deleted_at AS "deletedAt"
353 FROM projection_threads
354 WHERE deleted_at IS NULL
355 AND archived_at IS NULL
356 ORDER BY project_id ASC, created_at ASC, thread_id ASC
357 `,
358 });
359
360 const listArchivedThreadRows = SqlSchema.findAll({
361 Request: Schema.Void,
362 Result: ProjectionThreadDbRowSchema,
363 execute: () =>
364 sql`
365 SELECT
366 thread_id AS "threadId",
367 project_id AS "projectId",
368 title,
369 model_selection_json AS "modelSelection",
370 runtime_mode AS "runtimeMode",
371 interaction_mode AS "interactionMode",
372 branch,
373 worktree_path AS "worktreePath",
374 latest_turn_id AS "latestTurnId",
375 created_at AS "createdAt",
376 updated_at AS "updatedAt",
377 archived_at AS "archivedAt",
378 latest_user_message_at AS "latestUserMessageAt",
379 pending_approval_count AS "pendingApprovalCount",
380 pending_user_input_count AS "pendingUserInputCount",
381 has_actionable_proposed_plan AS "hasActionableProposedPlan",
382 deleted_at AS "deletedAt"
383 FROM projection_threads
384 WHERE deleted_at IS NULL
385 AND archived_at IS NOT NULL
386 ORDER BY project_id ASC, archived_at DESC, thread_id DESC
387 `,
388 });
389
330390 const listThreadMessageRows = SqlSchema.findAll({
331391 Request: Schema.Void,
332392 Result: ProjectionThreadMessageDbRowSchema,
......@@ -411,6 +471,56 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
411471 `,
412472 });
413473
474 const listActiveThreadSessionRows = SqlSchema.findAll({
475 Request: Schema.Void,
476 Result: ProjectionThreadSessionDbRowSchema,
477 execute: () =>
478 sql`
479 SELECT
480 sessions.thread_id AS "threadId",
481 sessions.status,
482 sessions.provider_name AS "providerName",
483 sessions.provider_instance_id AS "providerInstanceId",
484 sessions.provider_session_id AS "providerSessionId",
485 sessions.provider_thread_id AS "providerThreadId",
486 sessions.runtime_mode AS "runtimeMode",
487 sessions.active_turn_id AS "activeTurnId",
488 sessions.last_error AS "lastError",
489 sessions.updated_at AS "updatedAt"
490 FROM projection_thread_sessions sessions
491 INNER JOIN projection_threads threads
492 ON threads.thread_id = sessions.thread_id
493 WHERE threads.deleted_at IS NULL
494 AND threads.archived_at IS NULL
495 ORDER BY sessions.thread_id ASC
496 `,
497 });
498
499 const listArchivedThreadSessionRows = SqlSchema.findAll({
500 Request: Schema.Void,
501 Result: ProjectionThreadSessionDbRowSchema,
502 execute: () =>
503 sql`
504 SELECT
505 sessions.thread_id AS "threadId",
506 sessions.status,
507 sessions.provider_name AS "providerName",
508 sessions.provider_instance_id AS "providerInstanceId",
509 sessions.provider_session_id AS "providerSessionId",
510 sessions.provider_thread_id AS "providerThreadId",
511 sessions.runtime_mode AS "runtimeMode",
512 sessions.active_turn_id AS "activeTurnId",
513 sessions.last_error AS "lastError",
514 sessions.updated_at AS "updatedAt"
515 FROM projection_thread_sessions sessions
516 INNER JOIN projection_threads threads
517 ON threads.thread_id = sessions.thread_id
518 WHERE threads.deleted_at IS NULL
519 AND threads.archived_at IS NOT NULL
520 ORDER BY sessions.thread_id ASC
521 `,
522 });
523
414524 const listCheckpointRows = SqlSchema.findAll({
415525 Request: Schema.Void,
416526 Result: ProjectionCheckpointDbRowSchema,
......@@ -475,6 +585,33 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
475585 ON turns.thread_id = threads.thread_id
476586 AND turns.turn_id = threads.latest_turn_id
477587 WHERE threads.deleted_at IS NULL
588 AND threads.archived_at IS NULL
589 AND threads.latest_turn_id IS NOT NULL
590 ORDER BY turns.thread_id ASC
591 `,
592 });
593
594 const listArchivedLatestTurnRows = SqlSchema.findAll({
595 Request: Schema.Void,
596 Result: ProjectionLatestTurnDbRowSchema,
597 execute: () =>
598 sql`
599 SELECT
600 turns.thread_id AS "threadId",
601 turns.turn_id AS "turnId",
602 turns.state,
603 turns.requested_at AS "requestedAt",
604 turns.started_at AS "startedAt",
605 turns.completed_at AS "completedAt",
606 turns.assistant_message_id AS "assistantMessageId",
607 turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
608 turns.source_proposed_plan_id AS "sourceProposedPlanId"
609 FROM projection_threads threads
610 JOIN projection_turns turns
611 ON turns.thread_id = threads.thread_id
612 AND turns.turn_id = threads.latest_turn_id
613 WHERE threads.deleted_at IS NULL
614 AND threads.archived_at IS NOT NULL
478615 AND threads.latest_turn_id IS NOT NULL
479616 ORDER BY turns.thread_id ASC
480617 `,
......@@ -557,6 +694,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
557694 FROM projection_threads
558695 WHERE project_id = ${projectId}
559696 AND deleted_at IS NULL
697 AND archived_at IS NULL
560698 ORDER BY created_at ASC, thread_id ASC
561699 LIMIT 1
562700 `,
......@@ -607,6 +745,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
607745 FROM projection_threads
608746 WHERE thread_id = ${threadId}
609747 AND deleted_at IS NULL
748 AND archived_at IS NULL
610749 LIMIT 1
611750 `,
612751 });
......@@ -717,6 +856,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
717856 AND turns.turn_id = threads.latest_turn_id
718857 WHERE threads.thread_id = ${threadId}
719858 AND threads.deleted_at IS NULL
859 AND threads.archived_at IS NULL
720860 LIMIT 1
721861 `,
722862 });
......@@ -1232,7 +1372,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
12321372 ),
12331373 ),
12341374 ),
1235 listThreadRows(undefined).pipe(
1375 listActiveThreadRows(undefined).pipe(
12361376 Effect.mapError(
12371377 toPersistenceSqlOrDecodeError(
12381378 "ProjectionSnapshotQuery.getShellSnapshot:listThreads:query",
......@@ -1240,7 +1380,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
12401380 ),
12411381 ),
12421382 ),
1243 listThreadSessionRows(undefined).pipe(
1383 listActiveThreadSessionRows(undefined).pipe(
12441384 Effect.mapError(
12451385 toPersistenceSqlOrDecodeError(
12461386 "ProjectionSnapshotQuery.getShellSnapshot:listThreadSessions:query",
......@@ -1350,6 +1490,139 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
13501490 }),
13511491 );
13521492
1493 const getArchivedShellSnapshot: ProjectionSnapshotQueryShape["getArchivedShellSnapshot"] = () =>
1494 sql
1495 .withTransaction(
1496 Effect.all([
1497 listProjectRows(undefined).pipe(
1498 Effect.mapError(
1499 toPersistenceSqlOrDecodeError(
1500 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjects:query",
1501 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjects:decodeRows",
1502 ),
1503 ),
1504 ),
1505 listArchivedThreadRows(undefined).pipe(
1506 Effect.mapError(
1507 toPersistenceSqlOrDecodeError(
1508 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreads:query",
1509 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreads:decodeRows",
1510 ),
1511 ),
1512 ),
1513 listArchivedThreadSessionRows(undefined).pipe(
1514 Effect.mapError(
1515 toPersistenceSqlOrDecodeError(
1516 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreadSessions:query",
1517 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreadSessions:decodeRows",
1518 ),
1519 ),
1520 ),
1521 listArchivedLatestTurnRows(undefined).pipe(
1522 Effect.mapError(
1523 toPersistenceSqlOrDecodeError(
1524 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listLatestTurns:query",
1525 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listLatestTurns:decodeRows",
1526 ),
1527 ),
1528 ),
1529 listProjectionStateRows(undefined).pipe(
1530 Effect.mapError(
1531 toPersistenceSqlOrDecodeError(
1532 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjectionState:query",
1533 "ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjectionState:decodeRows",
1534 ),
1535 ),
1536 ),
1537 ]),
1538 )
1539 .pipe(
1540 Effect.flatMap(([projectRows, threadRows, sessionRows, latestTurnRows, stateRows]) =>
1541 Effect.gen(function* () {
1542 let updatedAt: string | null = null;
1543 for (const row of projectRows) {
1544 updatedAt = maxIso(updatedAt, row.updatedAt);
1545 }
1546 for (const row of threadRows) {
1547 updatedAt = maxIso(updatedAt, row.updatedAt);
1548 }
1549 for (const row of sessionRows) {
1550 updatedAt = maxIso(updatedAt, row.updatedAt);
1551 }
1552 for (const row of latestTurnRows) {
1553 updatedAt = maxIso(updatedAt, row.requestedAt);
1554 if (row.startedAt !== null) {
1555 updatedAt = maxIso(updatedAt, row.startedAt);
1556 }
1557 if (row.completedAt !== null) {
1558 updatedAt = maxIso(updatedAt, row.completedAt);
1559 }
1560 }
1561 for (const row of stateRows) {
1562 updatedAt = maxIso(updatedAt, row.updatedAt);
1563 }
1564
1565 const activeProjectIds = new Set(threadRows.map((row) => row.projectId));
1566 const repositoryIdentities = yield* resolveRepositoryIdentitiesForProjects(
1567 projectRows.filter((row) => activeProjectIds.has(row.projectId)),
1568 );
1569 const latestTurnByThread = new Map(
1570 latestTurnRows.map((row) => [row.threadId, mapLatestTurn(row)] as const),
1571 );
1572 const sessionByThread = new Map(
1573 sessionRows.map((row) => [row.threadId, mapSessionRow(row)] as const),
1574 );
1575
1576 const snapshot = {
1577 snapshotSequence: computeSnapshotSequence(stateRows),
1578 projects: projectRows
1579 .filter((row) => row.deletedAt === null && activeProjectIds.has(row.projectId))
1580 .map((row) =>
1581 mapProjectShellRow(row, repositoryIdentities.get(row.projectId) ?? null),
1582 ),
1583 threads: threadRows.map(
1584 (row): OrchestrationThreadShell => ({
1585 id: row.threadId,
1586 projectId: row.projectId,
1587 title: row.title,
1588 modelSelection: row.modelSelection,
1589 runtimeMode: row.runtimeMode,
1590 interactionMode: row.interactionMode,
1591 branch: row.branch,
1592 worktreePath: row.worktreePath,
1593 latestTurn: latestTurnByThread.get(row.threadId) ?? null,
1594 createdAt: row.createdAt,
1595 updatedAt: row.updatedAt,
1596 archivedAt: row.archivedAt,
1597 session: sessionByThread.get(row.threadId) ?? null,
1598 latestUserMessageAt: row.latestUserMessageAt,
1599 hasPendingApprovals: row.pendingApprovalCount > 0,
1600 hasPendingUserInput: row.pendingUserInputCount > 0,
1601 hasActionableProposedPlan: row.hasActionableProposedPlan > 0,
1602 }),
1603 ),
1604 updatedAt: updatedAt ?? new Date(0).toISOString(),
1605 };
1606
1607 return yield* decodeShellSnapshot(snapshot).pipe(
1608 Effect.mapError(
1609 toPersistenceDecodeError(
1610 "ProjectionSnapshotQuery.getArchivedShellSnapshot:decodeShellSnapshot",
1611 ),
1612 ),
1613 );
1614 }),
1615 ),
1616 Effect.mapError((error) => {
1617 if (isPersistenceError(error)) {
1618 return error;
1619 }
1620 return toPersistenceSqlError("ProjectionSnapshotQuery.getArchivedShellSnapshot:query")(
1621 error,
1622 );
1623 }),
1624 );
1625
13531626 const getSnapshotSequence: ProjectionSnapshotQueryShape["getSnapshotSequence"] = () =>
13541627 listProjectionStateRows(undefined).pipe(
13551628 Effect.mapError(
......@@ -1683,6 +1956,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
16831956 getCommandReadModel,
16841957 getSnapshot,
16851958 getShellSnapshot,
1959 getArchivedShellSnapshot,
16861960 getSnapshotSequence,
16871961 getCounts,
16881962 getActiveProjectByWorkspaceRoot,
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+11
......@@ -72,6 +72,17 @@ export interface ProjectionSnapshotQueryShape {
7272 ProjectionRepositoryError
7373 >;
7474
75 /**
76 * Read archived thread shell summaries for the archive page.
77 *
78 * This query is separate from the main shell snapshot so archived threads
79 * are never bootstrapped into normal navigation state.
80 */
81 readonly getArchivedShellSnapshot: () => Effect.Effect<
82 OrchestrationShellSnapshot,
83 ProjectionRepositoryError
84 >;
85
7586 /**
7687 * Read the latest projection snapshot sequence without hydrating read-model
7788 * entities.
apps/server/src/persistence/Migrations.ts+2
......@@ -42,6 +42,7 @@ import Migration0026 from "./Migrations/026_CanonicalizeModelSelectionOptions.ts
4242import Migration0027 from "./Migrations/027_ProviderSessionRuntimeInstanceId.ts";
4343import Migration0028 from "./Migrations/028_ProjectionThreadSessionInstanceId.ts";
4444import Migration0029 from "./Migrations/029_ProjectionThreadDetailOrderingIndexes.ts";
45import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes.ts";
4546
4647/**
4748 * Migration loader with all migrations defined inline.
......@@ -83,6 +84,7 @@ export const migrationEntries = [
8384 [27, "ProviderSessionRuntimeInstanceId", Migration0027],
8485 [28, "ProjectionThreadSessionInstanceId", Migration0028],
8586 [29, "ProjectionThreadDetailOrderingIndexes", Migration0029],
87 [30, "ProjectionThreadShellArchiveIndexes", Migration0030],
8688] as const;
8789
8890export const makeMigrationLoader = (throughId?: number) =>
apps/server/src/persistence/Migrations/030_ProjectionThreadShellArchiveIndexes.ts created+16
......@@ -0,0 +1,16 @@
1import * as Effect from "effect/Effect";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 yield* sql`
8 CREATE INDEX IF NOT EXISTS idx_projection_threads_shell_active
9 ON projection_threads(deleted_at, archived_at, project_id, created_at, thread_id)
10 `;
11
12 yield* sql`
13 CREATE INDEX IF NOT EXISTS idx_projection_threads_shell_archived
14 ON projection_threads(deleted_at, archived_at, project_id, thread_id)
15 `;
16});
apps/server/src/project/Layers/ProjectSetupScriptRunner.test.ts+1
......@@ -25,6 +25,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
2525 getCommandReadModel: () => Effect.die("unused"),
2626 getSnapshot: () => Effect.die("unused"),
2727 getShellSnapshot: () => Effect.die("unused"),
28 getArchivedShellSnapshot: () => Effect.die("unused"),
2829 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }),
2930 getCounts: () => Effect.die("unused"),
3031 getActiveProjectByWorkspaceRoot: (workspaceRoot) =>
apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+1
......@@ -194,6 +194,7 @@ describe("ProviderSessionReaper", () => {
194194 getCommandReadModel: () => Effect.die("unused"),
195195 getSnapshot: () => Effect.die("unused"),
196196 getShellSnapshot: () => Effect.die("unused"),
197 getArchivedShellSnapshot: () => Effect.die("unused"),
197198 getSnapshotSequence: () =>
198199 Effect.succeed({ snapshotSequence: input.readModel.snapshotSequence }),
199200 getCounts: () => Effect.die("unused"),
apps/server/src/server.test.ts+7
......@@ -630,6 +630,13 @@ const buildAppUnderTest = (options?: {
630630 threads: [],
631631 updatedAt: "1970-01-01T00:00:00.000Z",
632632 }),
633 getArchivedShellSnapshot: () =>
634 Effect.succeed({
635 snapshotSequence: 0,
636 projects: [],
637 threads: [],
638 updatedAt: new Date(0).toISOString(),
639 }),
633640 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
634641 getProjectShellById: () => Effect.succeed(Option.none()),
635642 getThreadShellById: () => Effect.succeed(Option.none()),
apps/server/src/serverRuntimeStartup.test.ts+3
......@@ -85,6 +85,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa
8585 getCommandReadModel: () => Effect.die("unused"),
8686 getSnapshot: () => Effect.die("unused"),
8787 getShellSnapshot: () => Effect.die("unused"),
88 getArchivedShellSnapshot: () => Effect.die("unused"),
8889 getSnapshotSequence: () => Effect.die("unused"),
8990 getCounts: () =>
9091 Deferred.await(releaseCounts).pipe(
......@@ -139,6 +140,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
139140 getCommandReadModel: () => Effect.die("unused"),
140141 getSnapshot: () => Effect.die("unused"),
141142 getShellSnapshot: () => Effect.die("unused"),
143 getArchivedShellSnapshot: () => Effect.die("unused"),
142144 getSnapshotSequence: () => Effect.die("unused"),
143145 getCounts: () => Effect.die("unused"),
144146 getActiveProjectByWorkspaceRoot: () =>
......@@ -191,6 +193,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when
191193 getCommandReadModel: () => Effect.die("unused"),
192194 getSnapshot: () => Effect.die("unused"),
193195 getShellSnapshot: () => Effect.die("unused"),
196 getArchivedShellSnapshot: () => Effect.die("unused"),
194197 getSnapshotSequence: () => Effect.die("unused"),
195198 getCounts: () => Effect.die("unused"),
196199 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
apps/server/src/ws.ts+29
......@@ -308,6 +308,7 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) =>
308308 }),
309309 );
310310 case "thread.deleted":
311 case "thread.archived":
311312 return Effect.succeed(
312313 Option.some({
313314 kind: "thread-removed" as const,
......@@ -315,6 +316,17 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) =>
315316 threadId: event.payload.threadId,
316317 }),
317318 );
319 case "thread.unarchived":
320 return projectionSnapshotQuery.getThreadShellById(event.payload.threadId).pipe(
321 Effect.map((thread) =>
322 Option.map(thread, (nextThread) => ({
323 kind: "thread-upserted" as const,
324 sequence: event.sequence,
325 thread: nextThread,
326 })),
327 ),
328 Effect.catch(() => Effect.succeed(Option.none())),
329 );
318330 default:
319331 if (event.aggregateKind !== "thread") {
320332 return Effect.succeed(Option.none());
......@@ -734,6 +746,23 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) =>
734746 }),
735747 { "rpc.aggregate": "orchestration" },
736748 ),
749 [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]: (_input) =>
750 observeRpcEffect(
751 ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot,
752 projectionSnapshotQuery.getArchivedShellSnapshot().pipe(
753 Effect.tapError((cause) =>
754 Effect.logError("orchestration archived shell snapshot load failed", { cause }),
755 ),
756 Effect.mapError(
757 (cause) =>
758 new OrchestrationGetSnapshotError({
759 message: "Failed to load archived orchestration shell snapshot",
760 cause,
761 }),
762 ),
763 ),
764 { "rpc.aggregate": "orchestration" },
765 ),
737766 [ORCHESTRATION_WS_METHODS.subscribeThread]: (input) =>
738767 observeRpcStreamEffect(
739768 ORCHESTRATION_WS_METHODS.subscribeThread,
apps/web/src/components/ChatView.browser.tsx+3
......@@ -218,6 +218,9 @@ function createMockEnvironmentApi(input: {
218218 getFullThreadDiff: (() => {
219219 throw new Error("Not implemented in browser test.");
220220 }) as EnvironmentApi["orchestration"]["getFullThreadDiff"],
221 getArchivedShellSnapshot: (() => {
222 throw new Error("Not implemented in browser test.");
223 }) as EnvironmentApi["orchestration"]["getArchivedShellSnapshot"],
221224 subscribeShell: (() => () => undefined) as EnvironmentApi["orchestration"]["subscribeShell"],
222225 subscribeThread: (() => () =>
223226 undefined) as EnvironmentApi["orchestration"]["subscribeThread"],
apps/web/src/components/settings/ConnectionsSettings.tsx+20-8
......@@ -2662,14 +2662,26 @@ export function ConnectionsSettings() {
26622662 }
26632663 }}
26642664 >
2665 <DialogTrigger
2666 render={
2667 <Button size="xs" variant="outline">
2668 <PlusIcon className="size-3" />
2669 Add environment
2670 </Button>
2671 }
2672 />
2665 <Tooltip>
2666 <TooltipTrigger
2667 render={
2668 <DialogTrigger
2669 render={
2670 <Button
2671 size="xs"
2672 variant="ghost"
2673 className="h-5 gap-1 rounded-sm px-1 text-[11px] font-normal text-muted-foreground/60 hover:text-muted-foreground"
2674 aria-label="Add environment"
2675 >
2676 <PlusIcon className="size-3" />
2677 <span>Add environment</span>
2678 </Button>
2679 }
2680 />
2681 }
2682 />
2683 <TooltipPopup side="top">Add environment</TooltipPopup>
2684 </Tooltip>
26732685 <DialogPopup className="max-h-[80dvh] sm:max-w-3xl">
26742686 <DialogHeader>
26752687 <DialogTitle>Add Environment</DialogTitle>
apps/web/src/components/settings/ProviderInstanceCard.tsx+1-1
......@@ -648,7 +648,7 @@ export function ProviderInstanceCard({
648648 );
649649
650650 const authRowNode = (
651 <p className="flex min-w-0 flex-wrap items-center gap-x-1 text-xs text-muted-foreground">
651 <p className="flex min-w-0 flex-wrap items-center gap-x-1 text-xs text-muted-foreground/80">
652652 {hasAuthenticatedEmail ? (
653653 <>
654654 <span>Authenticated as</span>
apps/web/src/components/settings/SettingsPanels.tsx+98-52
......@@ -44,14 +44,10 @@ import {
4444} from "../../providerInstances";
4545import { ensureLocalApi, readLocalApi } from "../../localApi";
4646import { useShallow } from "zustand/react/shallow";
47import {
48 selectProjectsAcrossEnvironments,
49 selectThreadShellsAcrossEnvironments,
50 useStore,
51} from "../../store";
47import { selectProjectsAcrossEnvironments, useStore } from "../../store";
48import { useArchivedThreadSnapshots } from "../../lib/archivedThreadsState";
5249import { formatRelativeTime, formatRelativeTimeLabel } from "../../timestampFormat";
5350import { Button } from "../ui/button";
54import { Empty, EmptyDescription, EmptyHeader, EmptyMedia, EmptyTitle } from "../ui/empty";
5551import { DraftInput } from "../ui/draft-input";
5652import { Select, SelectItem, SelectPopup, SelectTrigger, SelectValue } from "../ui/select";
5753import { Switch } from "../ui/switch";
......@@ -1330,14 +1326,50 @@ export function ProviderSettingsPanel() {
13301326
13311327export function ArchivedThreadsPanel() {
13321328 const projects = useStore(useShallow(selectProjectsAcrossEnvironments));
1333 const threads = useStore(useShallow(selectThreadShellsAcrossEnvironments));
13341329 const { unarchiveThread, confirmAndDeleteThread } = useThreadActions();
1330 const environmentIds = useMemo(
1331 () => [...new Set(projects.map((project) => project.environmentId))],
1332 [projects],
1333 );
1334 const {
1335 snapshots: archivedSnapshots,
1336 error: archiveError,
1337 isLoading: isLoadingArchive,
1338 refresh: refreshArchivedThreads,
1339 } = useArchivedThreadSnapshots(environmentIds);
1340
13351341 const archivedGroups = useMemo(() => {
1336 return projects
1342 const projectsByEnvironmentAndId = new Map(
1343 archivedSnapshots.flatMap(({ environmentId, snapshot }) =>
1344 snapshot.projects.map(
1345 (project) =>
1346 [
1347 `${environmentId}:${project.id}`,
1348 {
1349 id: project.id,
1350 environmentId,
1351 name: project.title,
1352 cwd: project.workspaceRoot,
1353 },
1354 ] as const,
1355 ),
1356 ),
1357 );
1358 const threads = archivedSnapshots.flatMap(({ environmentId, snapshot }) =>
1359 snapshot.threads.map((thread) => ({
1360 ...thread,
1361 environmentId,
1362 })),
1363 );
1364
1365 return [...projectsByEnvironmentAndId.values()]
13371366 .map((project) => ({
13381367 project,
13391368 threads: threads
1340 .filter((thread) => thread.projectId === project.id && thread.archivedAt !== null)
1369 .filter(
1370 (thread) =>
1371 thread.projectId === project.id && thread.environmentId === project.environmentId,
1372 )
13411373 .toSorted((left, right) => {
13421374 const leftKey = left.archivedAt ?? left.createdAt;
13431375 const rightKey = right.archivedAt ?? right.createdAt;
......@@ -1345,7 +1377,7 @@ export function ArchivedThreadsPanel() {
13451377 }),
13461378 }))
13471379 .filter((group) => group.threads.length > 0);
1348 }, [projects, threads]);
1380 }, [archivedSnapshots]);
13491381
13501382 const handleArchivedThreadContextMenu = useCallback(
13511383 async (threadRef: ScopedThreadRef, position: { x: number; y: number }) => {
......@@ -1362,6 +1394,7 @@ export function ArchivedThreadsPanel() {
13621394 if (clicked === "unarchive") {
13631395 try {
13641396 await unarchiveThread(threadRef);
1397 refreshArchivedThreads();
13651398 } catch (error) {
13661399 toastManager.add(
13671400 stackedThreadToast({
......@@ -1376,24 +1409,37 @@ export function ArchivedThreadsPanel() {
13761409
13771410 if (clicked === "delete") {
13781411 await confirmAndDeleteThread(threadRef);
1412 refreshArchivedThreads();
13791413 }
13801414 },
1381 [confirmAndDeleteThread, unarchiveThread],
1415 [confirmAndDeleteThread, refreshArchivedThreads, unarchiveThread],
13821416 );
13831417
13841418 return (
13851419 <SettingsPageContainer>
13861420 {archivedGroups.length === 0 ? (
13871421 <SettingsSection title="Archived threads">
1388 <Empty className="min-h-88">
1389 <EmptyMedia variant="icon">
1390 <ArchiveIcon />
1391 </EmptyMedia>
1392 <EmptyHeader>
1393 <EmptyTitle>No archived threads</EmptyTitle>
1394 <EmptyDescription>Archived threads will appear here.</EmptyDescription>
1395 </EmptyHeader>
1396 </Empty>
1422 <SettingsRow
1423 title={
1424 <span className="inline-flex items-center gap-2">
1425 {isLoadingArchive ? (
1426 <LoaderIcon className="size-3.5 animate-spin text-muted-foreground" />
1427 ) : (
1428 <ArchiveIcon className="size-3.5 text-muted-foreground" />
1429 )}
1430 {isLoadingArchive
1431 ? "Loading archived threads"
1432 : archiveError
1433 ? "Could not load archived threads"
1434 : "No archived threads"}
1435 </span>
1436 }
1437 description={
1438 isLoadingArchive
1439 ? "Checking connected environments."
1440 : (archiveError ?? "Archived threads will appear here.")
1441 }
1442 />
13971443 </SettingsSection>
13981444 ) : (
13991445 archivedGroups.map(({ project, threads: projectThreads }) => (
......@@ -1403,9 +1449,8 @@ export function ArchivedThreadsPanel() {
14031449 icon={<ProjectFavicon environmentId={project.environmentId} cwd={project.cwd} />}
14041450 >
14051451 {projectThreads.map((thread) => (
1406 <div
1452 <SettingsRow
14071453 key={thread.id}
1408 className="flex items-center justify-between gap-3 border-t border-border px-4 py-3 first:border-t-0 sm:px-5"
14091454 onContextMenu={(event) => {
14101455 event.preventDefault();
14111456 void handleArchivedThreadContextMenu(
......@@ -1416,39 +1461,40 @@ export function ArchivedThreadsPanel() {
14161461 },
14171462 );
14181463 }}
1419 >
1420 <div className="min-w-0 flex-1">
1421 <h3 className="truncate text-sm font-medium text-foreground">{thread.title}</h3>
1422 <p className="text-xs text-muted-foreground">
1464 title={thread.title}
1465 description={
1466 <>
14231467 Archived {formatRelativeTimeLabel(thread.archivedAt ?? thread.createdAt)}
14241468 {" \u00b7 Created "}
14251469 {formatRelativeTimeLabel(thread.createdAt)}
1426 </p>
1427 </div>
1428 <Button
1429 type="button"
1430 variant="outline"
1431 size="sm"
1432 className="h-7 shrink-0 cursor-pointer gap-1.5 px-2.5"
1433 onClick={() =>
1434 void unarchiveThread(scopeThreadRef(thread.environmentId, thread.id)).catch(
1435 (error) => {
1436 toastManager.add(
1437 stackedThreadToast({
1438 type: "error",
1439 title: "Failed to unarchive thread",
1440 description:
1441 error instanceof Error ? error.message : "An error occurred.",
1442 }),
1443 );
1444 },
1445 )
1446 }
1447 >
1448 <ArchiveX className="size-3.5" />
1449 <span>Unarchive</span>
1450 </Button>
1451 </div>
1470 </>
1471 }
1472 control={
1473 <Button
1474 type="button"
1475 variant="outline"
1476 size="sm"
1477 className="h-7 shrink-0 cursor-pointer gap-1.5 px-2.5"
1478 onClick={() =>
1479 void unarchiveThread(scopeThreadRef(thread.environmentId, thread.id))
1480 .then(() => refreshArchivedThreads())
1481 .catch((error) => {
1482 toastManager.add(
1483 stackedThreadToast({
1484 type: "error",
1485 title: "Failed to unarchive thread",
1486 description:
1487 error instanceof Error ? error.message : "An error occurred.",
1488 }),
1489 );
1490 })
1491 }
1492 >
1493 <ArchiveX className="size-3.5" />
1494 <span>Unarchive</span>
1495 </Button>
1496 }
1497 />
14521498 ))}
14531499 </SettingsSection>
14541500 ))
apps/web/src/components/settings/SourceControlSettings.tsx+1-1
......@@ -226,7 +226,7 @@ function DiscoveryItemRow({
226226 </Badge>
227227 ) : null}
228228 </div>
229 <p className="flex min-w-0 flex-wrap items-center gap-x-1 text-xs text-muted-foreground">
229 <p className="flex min-w-0 flex-wrap items-center gap-x-1 text-xs text-muted-foreground/80">
230230 {itemSummary({ item, auth, authAccount })}
231231 </p>
232232 </div>
apps/web/src/components/settings/settingsLayout.tsx+5-1
......@@ -52,7 +52,9 @@ export function SettingsRow({
5252 resetAction,
5353 control,
5454 children,
55}: {
55 className,
56 ...rowProps
57}: Omit<ComponentPropsWithoutRef<"div">, "title"> & {
5658 title: ReactNode;
5759 description: ReactNode;
5860 status?: ReactNode;
......@@ -62,9 +64,11 @@ export function SettingsRow({
6264}) {
6365 return (
6466 <div
67 {...rowProps}
6568 className={cn(
6669 "border-t border-border/60 px-4 first:border-t-0 sm:px-5",
6770 children ? "pt-3.5 pb-0" : "py-3.5",
71 className,
6872 )}
6973 >
7074 <div className="flex flex-col gap-3 sm:flex-row sm:items-center sm:justify-between">
apps/web/src/environmentApi.ts+1
......@@ -47,6 +47,7 @@ export function createEnvironmentApi(rpcClient: WsRpcClient): EnvironmentApi {
4747 dispatchCommand: rpcClient.orchestration.dispatchCommand,
4848 getTurnDiff: rpcClient.orchestration.getTurnDiff,
4949 getFullThreadDiff: rpcClient.orchestration.getFullThreadDiff,
50 getArchivedShellSnapshot: rpcClient.orchestration.getArchivedShellSnapshot,
5051 subscribeShell: (callback, options) =>
5152 rpcClient.orchestration.subscribeShell(callback, options),
5253 subscribeThread: (input, callback, options) =>
apps/web/src/hooks/useThreadActions.ts+24-10
......@@ -9,6 +9,7 @@ import { useComposerDraftStore } from "../composerDraftStore";
99import { useNewThreadHandler } from "./useHandleNewThread";
1010import { ensureEnvironmentApi, readEnvironmentApi } from "../environmentApi";
1111import { invalidateGitQueries } from "../lib/gitReactQuery";
12import { refreshArchivedThreadsForEnvironment } from "../lib/archivedThreadsState";
1213import { newCommandId } from "../lib/utils";
1314import { readLocalApi } from "../localApi";
1415import {
......@@ -68,19 +69,22 @@ export function useThreadActions() {
6869 throw new Error("Cannot archive a running thread.");
6970 }
7071
71 await api.orchestration.dispatchCommand({
72 const currentRouteThreadRef = getCurrentRouteThreadRef();
73 const shouldNavigateToDraft =
74 currentRouteThreadRef?.threadId === threadRef.threadId &&
75 currentRouteThreadRef.environmentId === threadRef.environmentId;
76 const archiveCommand = api.orchestration.dispatchCommand({
7277 type: "thread.archive",
7378 commandId: newCommandId(),
7479 threadId: threadRef.threadId,
7580 });
76 const currentRouteThreadRef = getCurrentRouteThreadRef();
7781
78 if (
79 currentRouteThreadRef?.threadId === threadRef.threadId &&
80 currentRouteThreadRef.environmentId === threadRef.environmentId
81 ) {
82 if (shouldNavigateToDraft) {
8283 await handleNewThreadRef.current(scopeProjectRef(thread.environmentId, thread.projectId));
8384 }
85
86 await archiveCommand;
87 refreshArchivedThreadsForEnvironment(threadRef.environmentId);
8488 },
8589 [getCurrentRouteThreadRef, resolveThreadTarget],
8690 );
......@@ -93,6 +97,7 @@ export function useThreadActions() {
9397 commandId: newCommandId(),
9498 threadId: target.threadId,
9599 });
100 refreshArchivedThreadsForEnvironment(target.environmentId);
96101 }, []);
97102
98103 const deleteThread = useCallback(
......@@ -100,7 +105,16 @@ export function useThreadActions() {
100105 const api = readEnvironmentApi(target.environmentId);
101106 if (!api) return;
102107 const resolved = resolveThreadTarget(target);
103 if (!resolved) return;
108 if (!resolved) {
109 // Thread not in main store (e.g. archived thread) — dispatch delete directly.
110 await api.orchestration.dispatchCommand({
111 type: "thread.delete",
112 commandId: newCommandId(),
113 threadId: target.threadId,
114 });
115 refreshArchivedThreadsForEnvironment(target.environmentId);
116 return;
117 }
104118 const { thread, threadRef } = resolved;
105119 const state = useStore.getState();
106120 const threads = selectThreadsForEnvironment(state, threadRef.environmentId);
......@@ -175,6 +189,7 @@ export function useThreadActions() {
175189 commandId: newCommandId(),
176190 threadId: threadRef.threadId,
177191 });
192 refreshArchivedThreadsForEnvironment(threadRef.environmentId);
178193 clearComposerDraftForThread(threadRef);
179194 clearProjectDraftThreadById(
180195 scopeProjectRef(threadRef.environmentId, thread.projectId),
......@@ -252,13 +267,12 @@ export function useThreadActions() {
252267 if (!api) return;
253268 const localApi = readLocalApi();
254269 const resolved = resolveThreadTarget(target);
255 if (!resolved) return;
256 const { thread } = resolved;
257270
258271 if (confirmThreadDelete && localApi) {
272 const title = resolved?.thread.title ?? "this thread";
259273 const confirmed = await localApi.dialogs.confirm(
260274 [
261 `Delete thread "${thread.title}"?`,
275 `Delete thread "${title}"?`,
262276 "This permanently clears conversation history for this thread.",
263277 ].join("\n"),
264278 );
apps/web/src/lib/archivedThreadsState.ts created+105
......@@ -0,0 +1,105 @@
1import { useAtomValue } from "@effect/atom-react";
2import { EnvironmentId, type OrchestrationShellSnapshot } from "@t3tools/contracts";
3import { Cause, Effect, Option } from "effect";
4import { AsyncResult, Atom } from "effect/unstable/reactivity";
5import { useCallback, useMemo } from "react";
6
7import { readEnvironmentApi } from "../environmentApi";
8import { appAtomRegistry } from "../rpc/atomRegistry";
9
10const ARCHIVED_THREADS_STALE_TIME_MS = 5_000;
11const ARCHIVED_THREADS_IDLE_TTL_MS = 5 * 60_000;
12const ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR = "\u001f";
13
14export type ArchivedSnapshotEntry = {
15 readonly environmentId: EnvironmentId;
16 readonly snapshot: OrchestrationShellSnapshot;
17};
18
19const knownArchivedThreadEnvironmentKeys = new Set<string>();
20
21function makeArchivedThreadsEnvironmentKey(environmentIds: ReadonlyArray<EnvironmentId>): string {
22 return environmentIds.toSorted().join(ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR);
23}
24
25function parseArchivedThreadsEnvironmentKey(key: string): ReadonlyArray<EnvironmentId> {
26 if (key.length === 0) {
27 return [];
28 }
29 return key
30 .split(ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR)
31 .map((environmentId) => EnvironmentId.make(environmentId));
32}
33
34const archivedThreadSnapshotsAtom = Atom.family((environmentKey: string) => {
35 knownArchivedThreadEnvironmentKeys.add(environmentKey);
36 return Atom.make(
37 Effect.promise(async (): Promise<ReadonlyArray<ArchivedSnapshotEntry>> => {
38 const environmentIds = parseArchivedThreadsEnvironmentKey(environmentKey);
39 const snapshots = await Promise.all(
40 environmentIds.map(async (environmentId) => {
41 const api = readEnvironmentApi(environmentId);
42 if (!api) {
43 return null;
44 }
45 return {
46 environmentId,
47 snapshot: await api.orchestration.getArchivedShellSnapshot(),
48 };
49 }),
50 );
51 return snapshots.filter((snapshot) => snapshot !== null);
52 }),
53 ).pipe(
54 Atom.swr({
55 staleTime: ARCHIVED_THREADS_STALE_TIME_MS,
56 revalidateOnMount: true,
57 }),
58 Atom.setIdleTTL(ARCHIVED_THREADS_IDLE_TTL_MS),
59 Atom.withLabel(`archived-thread-snapshots:${environmentKey}`),
60 );
61});
62
63function readArchivedThreadsError(
64 result: AsyncResult.AsyncResult<ReadonlyArray<ArchivedSnapshotEntry>, unknown>,
65): string | null {
66 if (result._tag !== "Failure") {
67 return null;
68 }
69
70 const error = Cause.squash(result.cause);
71 return error instanceof Error ? error.message : "Failed to load archived threads.";
72}
73
74export function refreshArchivedThreadsForEnvironment(environmentId: EnvironmentId): void {
75 for (const key of knownArchivedThreadEnvironmentKeys) {
76 if (parseArchivedThreadsEnvironmentKey(key).includes(environmentId)) {
77 appAtomRegistry.refresh(archivedThreadSnapshotsAtom(key));
78 }
79 }
80}
81
82export function useArchivedThreadSnapshots(environmentIds: ReadonlyArray<EnvironmentId>): {
83 readonly snapshots: ReadonlyArray<ArchivedSnapshotEntry>;
84 readonly error: string | null;
85 readonly isLoading: boolean;
86 readonly refresh: () => void;
87} {
88 const environmentKey = useMemo(
89 () => makeArchivedThreadsEnvironmentKey(environmentIds),
90 [environmentIds],
91 );
92 const atom = archivedThreadSnapshotsAtom(environmentKey);
93 const result = useAtomValue(atom);
94 const snapshots = Option.getOrElse(AsyncResult.value(result), () => []);
95 const refresh = useCallback(() => {
96 appAtomRegistry.refresh(atom);
97 }, [atom]);
98
99 return {
100 snapshots,
101 error: readArchivedThreadsError(result),
102 isLoading: result.waiting,
103 refresh,
104 };
105}
apps/web/src/rpc/wsRpcClient.ts+7
......@@ -143,6 +143,9 @@ export interface WsRpcClient {
143143 readonly dispatchCommand: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.dispatchCommand>;
144144 readonly getTurnDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getTurnDiff>;
145145 readonly getFullThreadDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getFullThreadDiff>;
146 readonly getArchivedShellSnapshot: RpcUnaryNoArgMethod<
147 typeof ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot
148 >;
146149 readonly subscribeShell: RpcStreamMethod<typeof ORCHESTRATION_WS_METHODS.subscribeShell>;
147150 readonly subscribeThread: RpcInputStreamMethod<typeof ORCHESTRATION_WS_METHODS.subscribeThread>;
148151 };
......@@ -288,6 +291,10 @@ export function createWsRpcClient(transport: WsTransport): WsRpcClient {
288291 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getTurnDiff](input)),
289292 getFullThreadDiff: (input) =>
290293 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getFullThreadDiff](input)),
294 getArchivedShellSnapshot: () =>
295 transport.request((client) =>
296 client[ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]({}),
297 ),
291298 subscribeShell: (listener, options) =>
292299 transport.subscribe(
293300 (client) => client[ORCHESTRATION_WS_METHODS.subscribeShell]({}),
packages/contracts/src/ipc.ts+2
......@@ -55,6 +55,7 @@ import type {
5555 OrchestrationGetFullThreadDiffResult,
5656 OrchestrationGetTurnDiffInput,
5757 OrchestrationGetTurnDiffResult,
58 OrchestrationShellSnapshot,
5859 OrchestrationShellStreamItem,
5960 OrchestrationSubscribeThreadInput,
6061 OrchestrationThreadStreamItem,
......@@ -544,6 +545,7 @@ export interface EnvironmentApi {
544545 getFullThreadDiff: (
545546 input: OrchestrationGetFullThreadDiffInput,
546547 ) => Promise<OrchestrationGetFullThreadDiffResult>;
548 getArchivedShellSnapshot: () => Promise<OrchestrationShellSnapshot>;
547549 subscribeShell: (
548550 callback: (event: OrchestrationShellStreamItem) => void,
549551 options?: {
packages/contracts/src/orchestration.ts+5
......@@ -27,6 +27,7 @@ export const ORCHESTRATION_WS_METHODS = {
2727 getTurnDiff: "orchestration.getTurnDiff",
2828 getFullThreadDiff: "orchestration.getFullThreadDiff",
2929 replayEvents: "orchestration.replayEvents",
30 getArchivedShellSnapshot: "orchestration.getArchivedShellSnapshot",
3031 subscribeShell: "orchestration.subscribeShell",
3132 subscribeThread: "orchestration.subscribeThread",
3233} as const;
......@@ -1222,6 +1223,10 @@ export const OrchestrationRpcSchemas = {
12221223 input: OrchestrationReplayEventsInput,
12231224 output: OrchestrationReplayEventsResult,
12241225 },
1226 getArchivedShellSnapshot: {
1227 input: Schema.Struct({}),
1228 output: OrchestrationShellSnapshot,
1229 },
12251230 subscribeThread: {
12261231 input: OrchestrationSubscribeThreadInput,
12271232 output: OrchestrationThreadStreamItem,
packages/contracts/src/rpc.ts+10
......@@ -410,6 +410,15 @@ export const WsOrchestrationReplayEventsRpc = Rpc.make(ORCHESTRATION_WS_METHODS.
410410 error: OrchestrationReplayEventsError,
411411});
412412
413export const WsOrchestrationGetArchivedShellSnapshotRpc = Rpc.make(
414 ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot,
415 {
416 payload: OrchestrationRpcSchemas.getArchivedShellSnapshot.input,
417 success: OrchestrationRpcSchemas.getArchivedShellSnapshot.output,
418 error: OrchestrationGetSnapshotError,
419 },
420);
421
413422export const WsOrchestrationSubscribeShellRpc = Rpc.make(ORCHESTRATION_WS_METHODS.subscribeShell, {
414423 payload: OrchestrationRpcSchemas.subscribeShell.input,
415424 success: OrchestrationRpcSchemas.subscribeShell.output,
......@@ -497,6 +506,7 @@ export const WsRpcGroup = RpcGroup.make(
497506 WsOrchestrationGetTurnDiffRpc,
498507 WsOrchestrationGetFullThreadDiffRpc,
499508 WsOrchestrationReplayEventsRpc,
509 WsOrchestrationGetArchivedShellSnapshotRpc,
500510 WsOrchestrationSubscribeShellRpc,
501511 WsOrchestrationSubscribeThreadRpc,
502512);