History

Move archived threads off the main read path

Julius Marminge committed 7 months ago

Commit bbe5e141 parent 393893021 files · +1393 −335Browse files

Unified · Split

21 files
  1. apps/server/integration/OrchestrationEngineHarness.integration.ts +6 −3
  2. apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts +6 −0
  3. apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +2 −0
  4. apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +258 −1
  5. apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +636 −242
  6. apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +18 −0
  7. apps/server/src/server.test.ts +29 −0
  8. apps/server/src/serverRuntimeStartup.test.ts +2 −0
  9. apps/server/src/ws.ts +22 −0
  10. apps/web/src/components/ChatView.browser.tsx +4 −1
  11. apps/web/src/components/KeybindingsToast.browser.tsx +4 −1
  12. apps/web/src/components/settings/SettingsPanels.tsx +123 −71
  13. apps/web/src/hooks/useThreadActions.ts +132 −11
  14. apps/web/src/lib/orchestrationReactQuery.ts +23 −0
  15. apps/web/src/routes/__root.tsx +36 −5
  16. apps/web/src/wsNativeApi.test.ts +2 −0
  17. apps/web/src/wsNativeApi.ts +2 −0
  18. apps/web/src/wsRpcClient.ts +10 −0
  19. packages/contracts/src/ipc.ts +3 −0
  20. packages/contracts/src/orchestration.ts +51 −0
  21. packages/contracts/src/rpc.ts +24 −0
apps/server/integration/OrchestrationEngineHarness.integration.ts+6−3
@@ -289,9 +289,10 @@ export const makeOrchestrationIntegrationHarness = ( · 288 unmodified lines
289 );289 );
290 290
291 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(GitCoreLive));291 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(GitCoreLive));
292 const projectionSnapshotQueryLayer = OrchestrationProjectionSnapshotQueryLive;
292 const runtimeServicesLayer = Layer.mergeAll(293 const runtimeServicesLayer = Layer.mergeAll(
293 orchestrationLayer,294 projectionSnapshotQueryLayer,
294 OrchestrationProjectionSnapshotQueryLive,295 orchestrationLayer.pipe(Layer.provide(projectionSnapshotQueryLayer)),
295 ProjectionCheckpointRepositoryLive,296 ProjectionCheckpointRepositoryLive,
296 ProjectionPendingApprovalRepositoryLive,297 ProjectionPendingApprovalRepositoryLive,
297 checkpointStoreLayer,298 checkpointStoreLayer,
@@ -333,7 +334,9 @@ export const makeOrchestrationIntegrationHarness = ( · 35 unmodified lines
333 Layer.provideMerge(providerCommandReactorLayer),334 Layer.provideMerge(providerCommandReactorLayer),
334 Layer.provideMerge(checkpointReactorLayer),335 Layer.provideMerge(checkpointReactorLayer),
335 );336 );
336 const layer = orchestrationReactorLayer.pipe(337 const layer = Layer.empty.pipe(
338 Layer.provideMerge(runtimeServicesLayer),
339 Layer.provideMerge(orchestrationReactorLayer),
337 Layer.provide(persistenceLayer),340 Layer.provide(persistenceLayer),
338 Layer.provideMerge(ServerSettingsService.layerTest()),341 Layer.provideMerge(ServerSettingsService.layerTest()),
339 Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),342 Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),
apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts+6−0
@@ -82,7 +82,10 @@ describe("CheckpointDiffQueryLive", () => { · 81 unmodified lines
82 Layer.succeed(ProjectionSnapshotQuery, {82 Layer.succeed(ProjectionSnapshotQuery, {
83 getSnapshot: () =>83 getSnapshot: () =>
84 Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),84 Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
85 getActiveSnapshot: () =>
86 Effect.die("CheckpointDiffQuery should not request the active orchestration snapshot"),
85 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),87 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
88 listArchivedThreads: () => Effect.succeed([]),
86 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),89 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
87 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),90 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
88 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),91 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
@@ -136,7 +139,10 @@ describe("CheckpointDiffQueryLive", () => { · 47 unmodified lines
136 Layer.succeed(ProjectionSnapshotQuery, {139 Layer.succeed(ProjectionSnapshotQuery, {
137 getSnapshot: () =>140 getSnapshot: () =>
138 Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),141 Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
142 getActiveSnapshot: () =>
143 Effect.die("CheckpointDiffQuery should not request the active orchestration snapshot"),
139 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),144 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
145 listArchivedThreads: () => Effect.succeed([]),
140 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),146 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
141 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),147 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
142 getThreadCheckpointContext: () => Effect.succeed(Option.none()),148 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+2−0
@@ -131,7 +131,9 @@ describe("OrchestrationEngine", () => { · 130 unmodified lines
131 Layer.provide(131 Layer.provide(
132 Layer.succeed(ProjectionSnapshotQuery, {132 Layer.succeed(ProjectionSnapshotQuery, {
133 getSnapshot: () => Effect.succeed(projectionSnapshot),133 getSnapshot: () => Effect.succeed(projectionSnapshot),
134 getActiveSnapshot: () => Effect.succeed(projectionSnapshot),
134 getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),135 getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
136 listArchivedThreads: () => Effect.succeed([]),
135 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),137 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
136 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),138 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
137 getThreadCheckpointContext: () => Effect.succeed(Option.none()),139 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+258−1
@@ -431,6 +431,21 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { · 430 unmodified lines
431 NULL,431 NULL,
432 NULL432 NULL
433 ),433 ),
434 (
435 'thread-archived',
436 'project-active',
437 'Archived Thread',
438 '{"provider":"codex","model":"gpt-5-codex"}',
439 'full-access',
440 'default',
441 NULL,
442 '/tmp/worktree-archived',
443 NULL,
444 '2026-03-01T00:00:04.000Z',
445 '2026-03-01T00:00:04.500Z',
446 '2026-03-01T00:00:04.750Z',
447 NULL
448 ),
434 (449 (
435 'thread-deleted',450 'thread-deleted',
436 'project-active',451 'project-active',
@@ -451,7 +466,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { · 14 unmodified lines
451 const counts = yield* snapshotQuery.getCounts();466 const counts = yield* snapshotQuery.getCounts();
452 assert.deepEqual(counts, {467 assert.deepEqual(counts, {
453 projectCount: 2,468 projectCount: 2,
454 threadCount: 3,469 threadCount: 4,
455 });470 });
456 471
457 const project = yield* snapshotQuery.getActiveProjectByWorkspaceRoot("/tmp/workspace");472 const project = yield* snapshotQuery.getActiveProjectByWorkspaceRoot("/tmp/workspace");
@@ -470,9 +485,251 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { · 12 unmodified lines
470 if (firstThreadId._tag === "Some") {485 if (firstThreadId._tag === "Some") {
471 assert.equal(firstThreadId.value, ThreadId.makeUnsafe("thread-first"));486 assert.equal(firstThreadId.value, ThreadId.makeUnsafe("thread-first"));
472 }487 }
488
489 const archivedThreads = yield* snapshotQuery.listArchivedThreads();
490 assert.deepEqual(archivedThreads, [
491 {
492 threadId: ThreadId.makeUnsafe("thread-archived"),
493 projectId: asProjectId("project-active"),
494 projectTitle: "Active Project",
495 workspaceRoot: "/tmp/workspace",
496 title: "Archived Thread",
497 worktreePath: "/tmp/worktree-archived",
498 createdAt: "2026-03-01T00:00:04.000Z",
499 updatedAt: "2026-03-01T00:00:04.500Z",
500 archivedAt: "2026-03-01T00:00:04.750Z",
501 },
502 ]);
473 }),503 }),
474 );504 );
475 505
506 it.effect("hydrates an active snapshot without archived or deleted thread data", () =>
507 Effect.gen(function* () {
508 const snapshotQuery = yield* ProjectionSnapshotQuery;
509 const sql = yield* SqlClient.SqlClient;
510
511 yield* sql`DELETE FROM projection_projects`;
512 yield* sql`DELETE FROM projection_threads`;
513 yield* sql`DELETE FROM projection_thread_messages`;
514 yield* sql`DELETE FROM projection_thread_proposed_plans`;
515 yield* sql`DELETE FROM projection_thread_activities`;
516 yield* sql`DELETE FROM projection_thread_sessions`;
517 yield* sql`DELETE FROM projection_turns`;
518 yield* sql`DELETE FROM projection_state`;
519
520 yield* sql`
521 INSERT INTO projection_projects (
522 project_id,
523 title,
524 workspace_root,
525 default_model_selection_json,
526 scripts_json,
527 created_at,
528 updated_at,
529 deleted_at
530 )
531 VALUES (
532 'project-active-only',
533 'Active-only Project',
534 '/tmp/active-only',
535 NULL,
536 '[]',
537 '2026-03-03T00:00:00.000Z',
538 '2026-03-03T00:00:01.000Z',
539 NULL
540 )
541 `;
542
543 yield* sql`
544 INSERT INTO projection_threads (
545 thread_id,
546 project_id,
547 title,
548 model_selection_json,
549 runtime_mode,
550 interaction_mode,
551 branch,
552 worktree_path,
553 latest_turn_id,
554 created_at,
555 updated_at,
556 archived_at,
557 deleted_at
558 )
559 VALUES
560 (
561 'thread-active-only',
562 'project-active-only',
563 'Active Thread',
564 '{"provider":"codex","model":"gpt-5-codex"}',
565 'full-access',
566 'default',
567 NULL,
568 NULL,
569 'turn-active-only',
570 '2026-03-03T00:00:02.000Z',
571 '2026-03-03T00:00:03.000Z',
572 NULL,
573 NULL
574 ),
575 (
576 'thread-archived-only',
577 'project-active-only',
578 'Archived Thread',
579 '{"provider":"codex","model":"gpt-5-codex"}',
580 'full-access',
581 'default',
582 NULL,
583 NULL,
584 'turn-archived-only',
585 '2026-03-03T00:00:04.000Z',
586 '2026-03-03T00:00:05.000Z',
587 '2026-03-03T00:00:06.000Z',
588 NULL
589 ),
590 (
591 'thread-deleted-only',
592 'project-active-only',
593 'Deleted Thread',
594 '{"provider":"codex","model":"gpt-5-codex"}',
595 'full-access',
596 'default',
597 NULL,
598 NULL,
599 NULL,
600 '2026-03-03T00:00:07.000Z',
601 '2026-03-03T00:00:08.000Z',
602 NULL,
603 '2026-03-03T00:00:09.000Z'
604 )
605 `;
606
607 yield* sql`
608 INSERT INTO projection_thread_messages (
609 message_id,
610 thread_id,
611 turn_id,
612 role,
613 text,
614 is_streaming,
615 created_at,
616 updated_at
617 )
618 VALUES
619 (
620 'message-active-only',
621 'thread-active-only',
622 'turn-active-only',
623 'assistant',
624 'active message',
625 0,
626 '2026-03-03T00:00:10.000Z',
627 '2026-03-03T00:00:11.000Z'
628 ),
629 (
630 'message-archived-only',
631 'thread-archived-only',
632 'turn-archived-only',
633 'assistant',
634 'archived message',
635 0,
636 '2026-03-03T00:00:12.000Z',
637 '2026-03-03T00:00:13.000Z'
638 )
639 `;
640
641 yield* sql`
642 INSERT INTO projection_turns (
643 thread_id,
644 turn_id,
645 pending_message_id,
646 source_proposed_plan_thread_id,
647 source_proposed_plan_id,
648 assistant_message_id,
649 state,
650 requested_at,
651 started_at,
652 completed_at,
653 checkpoint_turn_count,
654 checkpoint_ref,
655 checkpoint_status,
656 checkpoint_files_json
657 )
658 VALUES
659 (
660 'thread-active-only',
661 'turn-active-only',
662 NULL,
663 NULL,
664 NULL,
665 'message-active-only',
666 'completed',
667 '2026-03-03T00:00:14.000Z',
668 '2026-03-03T00:00:14.000Z',
669 '2026-03-03T00:00:14.000Z',
670 1,
671 'checkpoint-active-only',
672 'ready',
673 '[]'
674 ),
675 (
676 'thread-archived-only',
677 'turn-archived-only',
678 NULL,
679 NULL,
680 NULL,
681 'message-archived-only',
682 'completed',
683 '2026-03-03T00:00:15.000Z',
684 '2026-03-03T00:00:15.000Z',
685 '2026-03-03T00:00:15.000Z',
686 1,
687 'checkpoint-archived-only',
688 'ready',
689 '[]'
690 )
691 `;
692
693 let sequence = 9;
694 for (const projector of Object.values(ORCHESTRATION_PROJECTOR_NAMES)) {
695 yield* sql`
696 INSERT INTO projection_state (
697 projector,
698 last_applied_sequence,
699 updated_at
700 )
701 VALUES (
702 ${projector},
703 ${sequence},
704 '2026-03-03T00:00:16.000Z'
705 )
706 `;
707 sequence += 1;
708 }
709
710 const snapshot = yield* snapshotQuery.getActiveSnapshot();
711
712 assert.equal(snapshot.snapshotSequence, 9);
713 assert.equal(snapshot.projects.length, 1);
714 assert.deepEqual(
715 snapshot.threads.map((thread) => ({
716 id: thread.id,
717 title: thread.title,
718 messageCount: thread.messages.length,
719 checkpointCount: thread.checkpoints.length,
720 })),
721 [
722 {
723 id: ThreadId.makeUnsafe("thread-active-only"),
724 title: "Active Thread",
725 messageCount: 1,
726 checkpointCount: 1,
727 },
728 ],
729 );
730 }),
731 );
732
476 it.effect("reads single-thread checkpoint context without hydrating unrelated threads", () =>733 it.effect("reads single-thread checkpoint context without hydrating unrelated threads", () =>
477 Effect.gen(function* () {734 Effect.gen(function* () {
478 const snapshotQuery = yield* ProjectionSnapshotQuery;735 const snapshotQuery = yield* ProjectionSnapshotQuery;
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+636−242
@@ -3,6 +3,7 @@ import { · 2 unmodified lines
3 IsoDateTime,3 IsoDateTime,
4 MessageId,4 MessageId,
5 NonNegativeInt,5 NonNegativeInt,
6 OrchestrationArchivedThreadSummary,
6 OrchestrationCheckpointFile,7 OrchestrationCheckpointFile,
7 OrchestrationProposedPlanId,8 OrchestrationProposedPlanId,
8 OrchestrationReadModel,9 OrchestrationReadModel,
@@ -89,6 +90,7 @@ const ProjectionLatestTurnDbRowSchema = Schema.Struct({ · 80 unmodified lines
89 sourceProposedPlanId: Schema.NullOr(OrchestrationProposedPlanId),90 sourceProposedPlanId: Schema.NullOr(OrchestrationProposedPlanId),
90});91});
91const ProjectionStateDbRowSchema = ProjectionState;92const ProjectionStateDbRowSchema = ProjectionState;
93const ProjectionArchivedThreadSummaryRowSchema = OrchestrationArchivedThreadSummary;
92const ProjectionCountsRowSchema = Schema.Struct({94const ProjectionCountsRowSchema = Schema.Struct({
93 projectCount: Schema.Number,95 projectCount: Schema.Number,
94 threadCount: Schema.Number,96 threadCount: Schema.Number,
@@ -112,6 +114,17 @@ const ProjectionThreadCheckpointContextThreadRowSchema = Schema.Struct({ · 17 unmodified lines
112 workspaceRoot: Schema.String,114 workspaceRoot: Schema.String,
113 worktreePath: Schema.NullOr(Schema.String),115 worktreePath: Schema.NullOr(Schema.String),
114});116});
117type ProjectionProjectDbRow = Schema.Schema.Type<typeof ProjectionProjectDbRowSchema>;
118type ProjectionThreadDbRow = Schema.Schema.Type<typeof ProjectionThreadDbRowSchema>;
119type ProjectionThreadMessageDbRow = Schema.Schema.Type<typeof ProjectionThreadMessageDbRowSchema>;
120type ProjectionThreadProposedPlanDbRow = Schema.Schema.Type<
121 typeof ProjectionThreadProposedPlanDbRowSchema
122>;
123type ProjectionThreadActivityDbRow = Schema.Schema.Type<typeof ProjectionThreadActivityDbRowSchema>;
124type ProjectionThreadSessionDbRow = Schema.Schema.Type<typeof ProjectionThreadSessionDbRowSchema>;
125type ProjectionCheckpointDbRow = Schema.Schema.Type<typeof ProjectionCheckpointDbRowSchema>;
126type ProjectionLatestTurnDbRow = Schema.Schema.Type<typeof ProjectionLatestTurnDbRowSchema>;
127type ProjectionStateDbRow = Schema.Schema.Type<typeof ProjectionStateDbRowSchema>;
115 128
116const REQUIRED_SNAPSHOT_PROJECTORS = [129const REQUIRED_SNAPSHOT_PROJECTORS = [
117 ORCHESTRATION_PROJECTOR_NAMES.projects,130 ORCHESTRATION_PROJECTOR_NAMES.projects,
@@ -183,6 +196,26 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 65 unmodified lines
183 `,196 `,
184 });197 });
185 198
199 const listActiveProjectRows = SqlSchema.findAll({
200 Request: Schema.Void,
201 Result: ProjectionProjectDbRowSchema,
202 execute: () =>
203 sql`
204 SELECT
205 project_id AS "projectId",
206 title,
207 workspace_root AS "workspaceRoot",
208 default_model_selection_json AS "defaultModelSelection",
209 scripts_json AS "scripts",
210 created_at AS "createdAt",
211 updated_at AS "updatedAt",
212 deleted_at AS "deletedAt"
213 FROM projection_projects
214 WHERE deleted_at IS NULL
215 ORDER BY created_at ASC, project_id ASC
216 `,
217 });
218
186 const listThreadRows = SqlSchema.findAll({219 const listThreadRows = SqlSchema.findAll({
187 Request: Schema.Void,220 Request: Schema.Void,
188 Result: ProjectionThreadDbRowSchema,221 Result: ProjectionThreadDbRowSchema,
@@ -207,6 +240,32 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 18 unmodified lines
207 `,240 `,
208 });241 });
209 242
243 const listActiveThreadRows = SqlSchema.findAll({
244 Request: Schema.Void,
245 Result: ProjectionThreadDbRowSchema,
246 execute: () =>
247 sql`
248 SELECT
249 thread_id AS "threadId",
250 project_id AS "projectId",
251 title,
252 model_selection_json AS "modelSelection",
253 runtime_mode AS "runtimeMode",
254 interaction_mode AS "interactionMode",
255 branch,
256 worktree_path AS "worktreePath",
257 latest_turn_id AS "latestTurnId",
258 created_at AS "createdAt",
259 updated_at AS "updatedAt",
260 archived_at AS "archivedAt",
261 deleted_at AS "deletedAt"
262 FROM projection_threads
263 WHERE deleted_at IS NULL
264 AND archived_at IS NULL
265 ORDER BY created_at ASC, thread_id ASC
266 `,
267 });
268
210 const listThreadMessageRows = SqlSchema.findAll({269 const listThreadMessageRows = SqlSchema.findAll({
211 Request: Schema.Void,270 Request: Schema.Void,
212 Result: ProjectionThreadMessageDbRowSchema,271 Result: ProjectionThreadMessageDbRowSchema,
@@ -227,6 +286,32 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 14 unmodified lines
227 `,286 `,
228 });287 });
229 288
289 const listActiveThreadMessageRows = SqlSchema.findAll({
290 Request: Schema.Void,
291 Result: ProjectionThreadMessageDbRowSchema,
292 execute: () =>
293 sql`
294 SELECT
295 message_id AS "messageId",
296 thread_id AS "threadId",
297 turn_id AS "turnId",
298 role,
299 text,
300 attachments_json AS "attachments",
301 is_streaming AS "isStreaming",
302 created_at AS "createdAt",
303 updated_at AS "updatedAt"
304 FROM projection_thread_messages
305 WHERE thread_id IN (
306 SELECT thread_id
307 FROM projection_threads
308 WHERE deleted_at IS NULL
309 AND archived_at IS NULL
310 )
311 ORDER BY thread_id ASC, created_at ASC, message_id ASC
312 `,
313 });
314
230 const listThreadProposedPlanRows = SqlSchema.findAll({315 const listThreadProposedPlanRows = SqlSchema.findAll({
231 Request: Schema.Void,316 Request: Schema.Void,
232 Result: ProjectionThreadProposedPlanDbRowSchema,317 Result: ProjectionThreadProposedPlanDbRowSchema,
@@ -246,6 +331,31 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 13 unmodified lines
246 `,331 `,
247 });332 });
248 333
334 const listActiveThreadProposedPlanRows = SqlSchema.findAll({
335 Request: Schema.Void,
336 Result: ProjectionThreadProposedPlanDbRowSchema,
337 execute: () =>
338 sql`
339 SELECT
340 plan_id AS "planId",
341 thread_id AS "threadId",
342 turn_id AS "turnId",
343 plan_markdown AS "planMarkdown",
344 implemented_at AS "implementedAt",
345 implementation_thread_id AS "implementationThreadId",
346 created_at AS "createdAt",
347 updated_at AS "updatedAt"
348 FROM projection_thread_proposed_plans
349 WHERE thread_id IN (
350 SELECT thread_id
351 FROM projection_threads
352 WHERE deleted_at IS NULL
353 AND archived_at IS NULL
354 )
355 ORDER BY thread_id ASC, created_at ASC, plan_id ASC
356 `,
357 });
358
249 const listThreadActivityRows = SqlSchema.findAll({359 const listThreadActivityRows = SqlSchema.findAll({
250 Request: Schema.Void,360 Request: Schema.Void,
251 Result: ProjectionThreadActivityDbRowSchema,361 Result: ProjectionThreadActivityDbRowSchema,
@@ -271,6 +381,37 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 19 unmodified lines
271 `,381 `,
272 });382 });
273 383
384 const listActiveThreadActivityRows = SqlSchema.findAll({
385 Request: Schema.Void,
386 Result: ProjectionThreadActivityDbRowSchema,
387 execute: () =>
388 sql`
389 SELECT
390 activity_id AS "activityId",
391 thread_id AS "threadId",
392 turn_id AS "turnId",
393 tone,
394 kind,
395 summary,
396 payload_json AS "payload",
397 sequence,
398 created_at AS "createdAt"
399 FROM projection_thread_activities
400 WHERE thread_id IN (
401 SELECT thread_id
402 FROM projection_threads
403 WHERE deleted_at IS NULL
404 AND archived_at IS NULL
405 )
406 ORDER BY
407 thread_id ASC,
408 CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC,
409 sequence ASC,
410 created_at ASC,
411 activity_id ASC
412 `,
413 });
414
274 const listThreadSessionRows = SqlSchema.findAll({415 const listThreadSessionRows = SqlSchema.findAll({
275 Request: Schema.Void,416 Request: Schema.Void,
276 Result: ProjectionThreadSessionDbRowSchema,417 Result: ProjectionThreadSessionDbRowSchema,
@@ -291,6 +432,32 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 14 unmodified lines
291 `,432 `,
292 });433 });
293 434
435 const listActiveThreadSessionRows = SqlSchema.findAll({
436 Request: Schema.Void,
437 Result: ProjectionThreadSessionDbRowSchema,
438 execute: () =>
439 sql`
440 SELECT
441 thread_id AS "threadId",
442 status,
443 provider_name AS "providerName",
444 provider_session_id AS "providerSessionId",
445 provider_thread_id AS "providerThreadId",
446 runtime_mode AS "runtimeMode",
447 active_turn_id AS "activeTurnId",
448 last_error AS "lastError",
449 updated_at AS "updatedAt"
450 FROM projection_thread_sessions
451 WHERE thread_id IN (
452 SELECT thread_id
453 FROM projection_threads
454 WHERE deleted_at IS NULL
455 AND archived_at IS NULL
456 )
457 ORDER BY thread_id ASC
458 `,
459 });
460
294 const listCheckpointRows = SqlSchema.findAll({461 const listCheckpointRows = SqlSchema.findAll({
295 Request: Schema.Void,462 Request: Schema.Void,
296 Result: ProjectionCheckpointDbRowSchema,463 Result: ProjectionCheckpointDbRowSchema,
@@ -311,6 +478,32 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 14 unmodified lines
311 `,478 `,
312 });479 });
313 480
481 const listActiveCheckpointRows = SqlSchema.findAll({
482 Request: Schema.Void,
483 Result: ProjectionCheckpointDbRowSchema,
484 execute: () =>
485 sql`
486 SELECT
487 thread_id AS "threadId",
488 turn_id AS "turnId",
489 checkpoint_turn_count AS "checkpointTurnCount",
490 checkpoint_ref AS "checkpointRef",
491 checkpoint_status AS "status",
492 checkpoint_files_json AS "files",
493 assistant_message_id AS "assistantMessageId",
494 completed_at AS "completedAt"
495 FROM projection_turns
496 WHERE checkpoint_turn_count IS NOT NULL
497 AND thread_id IN (
498 SELECT thread_id
499 FROM projection_threads
500 WHERE deleted_at IS NULL
501 AND archived_at IS NULL
502 )
503 ORDER BY thread_id ASC, checkpoint_turn_count ASC
504 `,
505 });
506
314 const listLatestTurnRows = SqlSchema.findAll({507 const listLatestTurnRows = SqlSchema.findAll({
315 Request: Schema.Void,508 Request: Schema.Void,
316 Result: ProjectionLatestTurnDbRowSchema,509 Result: ProjectionLatestTurnDbRowSchema,
@@ -332,6 +525,33 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 15 unmodified lines
332 `,525 `,
333 });526 });
334 527
528 const listActiveLatestTurnRows = SqlSchema.findAll({
529 Request: Schema.Void,
530 Result: ProjectionLatestTurnDbRowSchema,
531 execute: () =>
532 sql`
533 SELECT
534 thread_id AS "threadId",
535 turn_id AS "turnId",
536 state,
537 requested_at AS "requestedAt",
538 started_at AS "startedAt",
539 completed_at AS "completedAt",
540 assistant_message_id AS "assistantMessageId",
541 source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
542 source_proposed_plan_id AS "sourceProposedPlanId"
543 FROM projection_turns
544 WHERE turn_id IS NOT NULL
545 AND thread_id IN (
546 SELECT thread_id
547 FROM projection_threads
548 WHERE deleted_at IS NULL
549 AND archived_at IS NULL
550 )
551 ORDER BY thread_id ASC, requested_at DESC, turn_id DESC
552 `,
553 });
554
335 const listProjectionStateRows = SqlSchema.findAll({555 const listProjectionStateRows = SqlSchema.findAll({
336 Request: Schema.Void,556 Request: Schema.Void,
337 Result: ProjectionStateDbRowSchema,557 Result: ProjectionStateDbRowSchema,
@@ -345,6 +565,31 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 7 unmodified lines
345 `,565 `,
346 });566 });
347 567
568 const listArchivedThreadSummaryRows = SqlSchema.findAll({
569 Request: Schema.Void,
570 Result: ProjectionArchivedThreadSummaryRowSchema,
571 execute: () =>
572 sql`
573 SELECT
574 threads.thread_id AS "threadId",
575 threads.project_id AS "projectId",
576 projects.title AS "projectTitle",
577 projects.workspace_root AS "workspaceRoot",
578 threads.title,
579 threads.worktree_path AS "worktreePath",
580 threads.created_at AS "createdAt",
581 threads.updated_at AS "updatedAt",
582 threads.archived_at AS "archivedAt"
583 FROM projection_threads AS threads
584 INNER JOIN projection_projects AS projects
585 ON projects.project_id = threads.project_id
586 WHERE threads.deleted_at IS NULL
587 AND threads.archived_at IS NOT NULL
588 AND projects.deleted_at IS NULL
589 ORDER BY threads.archived_at DESC, threads.thread_id DESC
590 `,
591 });
592
348 const readProjectionCounts = SqlSchema.findOne({593 const readProjectionCounts = SqlSchema.findOne({
349 Request: Schema.Void,594 Request: Schema.Void,
350 Result: ProjectionCountsRowSchema,595 Result: ProjectionCountsRowSchema,
@@ -388,6 +633,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 37 unmodified lines
388 FROM projection_threads633 FROM projection_threads
389 WHERE project_id = ${projectId}634 WHERE project_id = ${projectId}
390 AND deleted_at IS NULL635 AND deleted_at IS NULL
636 AND archived_at IS NULL
391 ORDER BY created_at ASC, thread_id ASC637 ORDER BY created_at ASC, thread_id ASC
392 LIMIT 1638 LIMIT 1
393 `,639 `,
@@ -433,7 +679,223 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 39 unmodified lines
433 `,679 `,
434 });680 });
435 681
436 const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () =>682 const buildSnapshot = (input: {
683 readonly operationPrefix: string;
684 readonly projectRows: ReadonlyArray<ProjectionProjectDbRow>;
685 readonly threadRows: ReadonlyArray<ProjectionThreadDbRow>;
686 readonly messageRows: ReadonlyArray<ProjectionThreadMessageDbRow>;
687 readonly proposedPlanRows: ReadonlyArray<ProjectionThreadProposedPlanDbRow>;
688 readonly activityRows: ReadonlyArray<ProjectionThreadActivityDbRow>;
689 readonly sessionRows: ReadonlyArray<ProjectionThreadSessionDbRow>;
690 readonly checkpointRows: ReadonlyArray<ProjectionCheckpointDbRow>;
691 readonly latestTurnRows: ReadonlyArray<ProjectionLatestTurnDbRow>;
692 readonly stateRows: ReadonlyArray<ProjectionStateDbRow>;
693 }) => {
694 const messagesByThread = new Map<string, Array<OrchestrationMessage>>();
695 const proposedPlansByThread = new Map<string, Array<OrchestrationProposedPlan>>();
696 const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();
697 const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();
698 const sessionsByThread = new Map<string, OrchestrationSession>();
699 const latestTurnByThread = new Map<string, OrchestrationLatestTurn>();
700
701 let updatedAt: string | null = null;
702
703 for (const row of input.projectRows) {
704 updatedAt = maxIso(updatedAt, row.updatedAt);
705 }
706 for (const row of input.threadRows) {
707 updatedAt = maxIso(updatedAt, row.updatedAt);
708 }
709 for (const row of input.stateRows) {
710 updatedAt = maxIso(updatedAt, row.updatedAt);
711 }
712
713 for (const row of input.messageRows) {
714 updatedAt = maxIso(updatedAt, row.updatedAt);
715 const threadMessages = messagesByThread.get(row.threadId) ?? [];
716 threadMessages.push({
717 id: row.messageId,
718 role: row.role,
719 text: row.text,
720 ...(row.attachments !== null ? { attachments: row.attachments } : {}),
721 turnId: row.turnId,
722 streaming: row.isStreaming === 1,
723 createdAt: row.createdAt,
724 updatedAt: row.updatedAt,
725 });
726 messagesByThread.set(row.threadId, threadMessages);
727 }
728
729 for (const row of input.proposedPlanRows) {
730 updatedAt = maxIso(updatedAt, row.updatedAt);
731 const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? [];
732 threadProposedPlans.push({
733 id: row.planId,
734 turnId: row.turnId,
735 planMarkdown: row.planMarkdown,
736 implementedAt: row.implementedAt,
737 implementationThreadId: row.implementationThreadId,
738 createdAt: row.createdAt,
739 updatedAt: row.updatedAt,
740 });
741 proposedPlansByThread.set(row.threadId, threadProposedPlans);
742 }
743
744 for (const row of input.activityRows) {
745 updatedAt = maxIso(updatedAt, row.createdAt);
746 const threadActivities = activitiesByThread.get(row.threadId) ?? [];
747 threadActivities.push({
748 id: row.activityId,
749 tone: row.tone,
750 kind: row.kind,
751 summary: row.summary,
752 payload: row.payload,
753 turnId: row.turnId,
754 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
755 createdAt: row.createdAt,
756 });
757 activitiesByThread.set(row.threadId, threadActivities);
758 }
759
760 for (const row of input.checkpointRows) {
761 updatedAt = maxIso(updatedAt, row.completedAt);
762 const threadCheckpoints = checkpointsByThread.get(row.threadId) ?? [];
763 threadCheckpoints.push({
764 turnId: row.turnId,
765 checkpointTurnCount: row.checkpointTurnCount,
766 checkpointRef: row.checkpointRef,
767 status: row.status,
768 files: row.files,
769 assistantMessageId: row.assistantMessageId,
770 completedAt: row.completedAt,
771 });
772 checkpointsByThread.set(row.threadId, threadCheckpoints);
773 }
774
775 for (const row of input.latestTurnRows) {
776 updatedAt = maxIso(updatedAt, row.requestedAt);
777 if (row.startedAt !== null) {
778 updatedAt = maxIso(updatedAt, row.startedAt);
779 }
780 if (row.completedAt !== null) {
781 updatedAt = maxIso(updatedAt, row.completedAt);
782 }
783 if (latestTurnByThread.has(row.threadId)) {
784 continue;
785 }
786 latestTurnByThread.set(row.threadId, {
787 turnId: row.turnId,
788 state:
789 row.state === "error"
790 ? "error"
791 : row.state === "interrupted"
792 ? "interrupted"
793 : row.state === "completed"
794 ? "completed"
795 : "running",
796 requestedAt: row.requestedAt,
797 startedAt: row.startedAt,
798 completedAt: row.completedAt,
799 assistantMessageId: row.assistantMessageId,
800 ...(row.sourceProposedPlanThreadId !== null && row.sourceProposedPlanId !== null
801 ? {
802 sourceProposedPlan: {
803 threadId: row.sourceProposedPlanThreadId,
804 planId: row.sourceProposedPlanId,
805 },
806 }
807 : {}),
808 });
809 }
810
811 for (const row of input.sessionRows) {
812 updatedAt = maxIso(updatedAt, row.updatedAt);
813 sessionsByThread.set(row.threadId, {
814 threadId: row.threadId,
815 status: row.status,
816 providerName: row.providerName,
817 runtimeMode: row.runtimeMode,
818 activeTurnId: row.activeTurnId,
819 lastError: row.lastError,
820 updatedAt: row.updatedAt,
821 });
822 }
823
824 const projects: ReadonlyArray<OrchestrationProject> = input.projectRows.map((row) => ({
825 id: row.projectId,
826 title: row.title,
827 workspaceRoot: row.workspaceRoot,
828 defaultModelSelection: row.defaultModelSelection,
829 scripts: row.scripts,
830 createdAt: row.createdAt,
831 updatedAt: row.updatedAt,
832 deletedAt: row.deletedAt,
833 }));
834
835 const threads: ReadonlyArray<OrchestrationThread> = input.threadRows.map((row) => ({
836 id: row.threadId,
837 projectId: row.projectId,
838 title: row.title,
839 modelSelection: row.modelSelection,
840 runtimeMode: row.runtimeMode,
841 interactionMode: row.interactionMode,
842 branch: row.branch,
843 worktreePath: row.worktreePath,
844 latestTurn: latestTurnByThread.get(row.threadId) ?? null,
845 createdAt: row.createdAt,
846 updatedAt: row.updatedAt,
847 archivedAt: row.archivedAt,
848 deletedAt: row.deletedAt,
849 messages: messagesByThread.get(row.threadId) ?? [],
850 proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
851 activities: activitiesByThread.get(row.threadId) ?? [],
852 checkpoints: checkpointsByThread.get(row.threadId) ?? [],
853 session: sessionsByThread.get(row.threadId) ?? null,
854 }));
855
856 return decodeReadModel({
857 snapshotSequence: computeSnapshotSequence(input.stateRows),
858 projects,
859 threads,
860 updatedAt: updatedAt ?? new Date(0).toISOString(),
861 }).pipe(Effect.mapError(toPersistenceDecodeError(`${input.operationPrefix}:decodeReadModel`)));
862 };
863
864 const getSnapshotFromQueries = (input: {
865 readonly operationPrefix: string;
866 readonly loadProjectRows: Effect.Effect<
867 ReadonlyArray<ProjectionProjectDbRow>,
868 ProjectionRepositoryError
869 >;
870 readonly loadThreadRows: Effect.Effect<
871 ReadonlyArray<ProjectionThreadDbRow>,
872 ProjectionRepositoryError
873 >;
874 readonly loadMessageRows: Effect.Effect<
875 ReadonlyArray<ProjectionThreadMessageDbRow>,
876 ProjectionRepositoryError
877 >;
878 readonly loadProposedPlanRows: Effect.Effect<
879 ReadonlyArray<ProjectionThreadProposedPlanDbRow>,
880 ProjectionRepositoryError
881 >;
882 readonly loadActivityRows: Effect.Effect<
883 ReadonlyArray<ProjectionThreadActivityDbRow>,
884 ProjectionRepositoryError
885 >;
886 readonly loadSessionRows: Effect.Effect<
887 ReadonlyArray<ProjectionThreadSessionDbRow>,
888 ProjectionRepositoryError
889 >;
890 readonly loadCheckpointRows: Effect.Effect<
891 ReadonlyArray<ProjectionCheckpointDbRow>,
892 ProjectionRepositoryError
893 >;
894 readonly loadLatestTurnRows: Effect.Effect<
895 ReadonlyArray<ProjectionLatestTurnDbRow>,
896 ProjectionRepositoryError
897 >;
898 }) =>
437 sql899 sql
438 .withTransaction(900 .withTransaction(
439 Effect.gen(function* () {901 Effect.gen(function* () {
@@ -448,254 +910,36 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 8 unmodified lines
448 latestTurnRows,910 latestTurnRows,
449 stateRows,911 stateRows,
450 ] = yield* Effect.all([912 ] = yield* Effect.all([
451 listProjectRows(undefined).pipe(913 input.loadProjectRows,
452 Effect.mapError(914 input.loadThreadRows,
453 toPersistenceSqlOrDecodeError(915 input.loadMessageRows,
454 "ProjectionSnapshotQuery.getSnapshot:listProjects:query",916 input.loadProposedPlanRows,
455 "ProjectionSnapshotQuery.getSnapshot:listProjects:decodeRows",917 input.loadActivityRows,
456 ),918 input.loadSessionRows,
457 ),919 input.loadCheckpointRows,
458 ),920 input.loadLatestTurnRows,
459 listThreadRows(undefined).pipe(
460 Effect.mapError(
461 toPersistenceSqlOrDecodeError(
462 "ProjectionSnapshotQuery.getSnapshot:listThreads:query",
463 "ProjectionSnapshotQuery.getSnapshot:listThreads:decodeRows",
464 ),
465 ),
466 ),
467 listThreadMessageRows(undefined).pipe(
468 Effect.mapError(
469 toPersistenceSqlOrDecodeError(
470 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:query",
471 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:decodeRows",
472 ),
473 ),
474 ),
475 listThreadProposedPlanRows(undefined).pipe(
476 Effect.mapError(
477 toPersistenceSqlOrDecodeError(
478 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:query",
479 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:decodeRows",
480 ),
481 ),
482 ),
483 listThreadActivityRows(undefined).pipe(
484 Effect.mapError(
485 toPersistenceSqlOrDecodeError(
486 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:query",
487 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:decodeRows",
488 ),
489 ),
490 ),
491 listThreadSessionRows(undefined).pipe(
492 Effect.mapError(
493 toPersistenceSqlOrDecodeError(
494 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:query",
495 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:decodeRows",
496 ),
497 ),
498 ),
499 listCheckpointRows(undefined).pipe(
500 Effect.mapError(
501 toPersistenceSqlOrDecodeError(
502 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:query",
503 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:decodeRows",
504 ),
505 ),
506 ),
507 listLatestTurnRows(undefined).pipe(
508 Effect.mapError(
509 toPersistenceSqlOrDecodeError(
510 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:query",
511 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:decodeRows",
512 ),
513 ),
514 ),
515 listProjectionStateRows(undefined).pipe(921 listProjectionStateRows(undefined).pipe(
516 Effect.mapError(922 Effect.mapError(
517 toPersistenceSqlOrDecodeError(923 toPersistenceSqlOrDecodeError(
518 "ProjectionSnapshotQuery.getSnapshot:listProjectionState:query",924 `${input.operationPrefix}:listProjectionState:query`,
519 "ProjectionSnapshotQuery.getSnapshot:listProjectionState:decodeRows",925 `${input.operationPrefix}:listProjectionState:decodeRows`,
520 ),926 ),
521 ),927 ),
522 ),928 ),
523 ]);929 ]);
524 930
525 const messagesByThread = new Map<string, Array<OrchestrationMessage>>();931 return yield* buildSnapshot({
526 const proposedPlansByThread = new Map<string, Array<OrchestrationProposedPlan>>();932 operationPrefix: input.operationPrefix,
527 const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();933 projectRows,
528 const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();934 threadRows,
529 const sessionsByThread = new Map<string, OrchestrationSession>();935 messageRows,
530 const latestTurnByThread = new Map<string, OrchestrationLatestTurn>();936 proposedPlanRows,
531 937 activityRows,
532 let updatedAt: string | null = null;938 sessionRows,
533 939 checkpointRows,
534 for (const row of projectRows) {940 latestTurnRows,
535 updatedAt = maxIso(updatedAt, row.updatedAt);941 stateRows,
536 }942 });
537 for (const row of threadRows) {
538 updatedAt = maxIso(updatedAt, row.updatedAt);
539 }
540 for (const row of stateRows) {
541 updatedAt = maxIso(updatedAt, row.updatedAt);
542 }
543
544 for (const row of messageRows) {
545 updatedAt = maxIso(updatedAt, row.updatedAt);
546 const threadMessages = messagesByThread.get(row.threadId) ?? [];
547 threadMessages.push({
548 id: row.messageId,
549 role: row.role,
550 text: row.text,
551 ...(row.attachments !== null ? { attachments: row.attachments } : {}),
552 turnId: row.turnId,
553 streaming: row.isStreaming === 1,
554 createdAt: row.createdAt,
555 updatedAt: row.updatedAt,
556 });
557 messagesByThread.set(row.threadId, threadMessages);
558 }
559
560 for (const row of proposedPlanRows) {
561 updatedAt = maxIso(updatedAt, row.updatedAt);
562 const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? [];
563 threadProposedPlans.push({
564 id: row.planId,
565 turnId: row.turnId,
566 planMarkdown: row.planMarkdown,
567 implementedAt: row.implementedAt,
568 implementationThreadId: row.implementationThreadId,
569 createdAt: row.createdAt,
570 updatedAt: row.updatedAt,
571 });
572 proposedPlansByThread.set(row.threadId, threadProposedPlans);
573 }
574
575 for (const row of activityRows) {
576 updatedAt = maxIso(updatedAt, row.createdAt);
577 const threadActivities = activitiesByThread.get(row.threadId) ?? [];
578 threadActivities.push({
579 id: row.activityId,
580 tone: row.tone,
581 kind: row.kind,
582 summary: row.summary,
583 payload: row.payload,
584 turnId: row.turnId,
585 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
586 createdAt: row.createdAt,
587 });
588 activitiesByThread.set(row.threadId, threadActivities);
589 }
590
591 for (const row of checkpointRows) {
592 updatedAt = maxIso(updatedAt, row.completedAt);
593 const threadCheckpoints = checkpointsByThread.get(row.threadId) ?? [];
594 threadCheckpoints.push({
595 turnId: row.turnId,
596 checkpointTurnCount: row.checkpointTurnCount,
597 checkpointRef: row.checkpointRef,
598 status: row.status,
599 files: row.files,
600 assistantMessageId: row.assistantMessageId,
601 completedAt: row.completedAt,
602 });
603 checkpointsByThread.set(row.threadId, threadCheckpoints);
604 }
605
606 for (const row of latestTurnRows) {
607 updatedAt = maxIso(updatedAt, row.requestedAt);
608 if (row.startedAt !== null) {
609 updatedAt = maxIso(updatedAt, row.startedAt);
610 }
611 if (row.completedAt !== null) {
612 updatedAt = maxIso(updatedAt, row.completedAt);
613 }
614 if (latestTurnByThread.has(row.threadId)) {
615 continue;
616 }
617 latestTurnByThread.set(row.threadId, {
618 turnId: row.turnId,
619 state:
620 row.state === "error"
621 ? "error"
622 : row.state === "interrupted"
623 ? "interrupted"
624 : row.state === "completed"
625 ? "completed"
626 : "running",
627 requestedAt: row.requestedAt,
628 startedAt: row.startedAt,
629 completedAt: row.completedAt,
630 assistantMessageId: row.assistantMessageId,
631 ...(row.sourceProposedPlanThreadId !== null && row.sourceProposedPlanId !== null
632 ? {
633 sourceProposedPlan: {
634 threadId: row.sourceProposedPlanThreadId,
635 planId: row.sourceProposedPlanId,
636 },
637 }
638 : {}),
639 });
640 }
641
642 for (const row of sessionRows) {
643 updatedAt = maxIso(updatedAt, row.updatedAt);
644 sessionsByThread.set(row.threadId, {
645 threadId: row.threadId,
646 status: row.status,
647 providerName: row.providerName,
648 runtimeMode: row.runtimeMode,
649 activeTurnId: row.activeTurnId,
650 lastError: row.lastError,
651 updatedAt: row.updatedAt,
652 });
653 }
654
655 const projects: ReadonlyArray<OrchestrationProject> = projectRows.map((row) => ({
656 id: row.projectId,
657 title: row.title,
658 workspaceRoot: row.workspaceRoot,
659 defaultModelSelection: row.defaultModelSelection,
660 scripts: row.scripts,
661 createdAt: row.createdAt,
662 updatedAt: row.updatedAt,
663 deletedAt: row.deletedAt,
664 }));
665
666 const threads: ReadonlyArray<OrchestrationThread> = threadRows.map((row) => ({
667 id: row.threadId,
668 projectId: row.projectId,
669 title: row.title,
670 modelSelection: row.modelSelection,
671 runtimeMode: row.runtimeMode,
672 interactionMode: row.interactionMode,
673 branch: row.branch,
674 worktreePath: row.worktreePath,
675 latestTurn: latestTurnByThread.get(row.threadId) ?? null,
676 createdAt: row.createdAt,
677 updatedAt: row.updatedAt,
678 archivedAt: row.archivedAt,
679 deletedAt: row.deletedAt,
680 messages: messagesByThread.get(row.threadId) ?? [],
681 proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
682 activities: activitiesByThread.get(row.threadId) ?? [],
683 checkpoints: checkpointsByThread.get(row.threadId) ?? [],
684 session: sessionsByThread.get(row.threadId) ?? null,
685 }));
686
687 const snapshot = {
688 snapshotSequence: computeSnapshotSequence(stateRows),
689 projects,
690 threads,
691 updatedAt: updatedAt ?? new Date(0).toISOString(),
692 };
693
694 return yield* decodeReadModel(snapshot).pipe(
695 Effect.mapError(
696 toPersistenceDecodeError("ProjectionSnapshotQuery.getSnapshot:decodeReadModel"),
697 ),
698 );
699 }),943 }),
700 )944 )
701 .pipe(945 .pipe(
@@ -703,10 +947,148 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 1 unmodified line
703 if (isPersistenceError(error)) {947 if (isPersistenceError(error)) {
704 return error;948 return error;
705 }949 }
706 return toPersistenceSqlError("ProjectionSnapshotQuery.getSnapshot:query")(error);950 return toPersistenceSqlError(`${input.operationPrefix}:query`)(error);
707 }),951 }),
708 );952 );
709 953
954 const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () =>
955 getSnapshotFromQueries({
956 operationPrefix: "ProjectionSnapshotQuery.getSnapshot",
957 loadProjectRows: listProjectRows(undefined).pipe(
958 Effect.mapError(
959 toPersistenceSqlOrDecodeError(
960 "ProjectionSnapshotQuery.getSnapshot:listProjects:query",
961 "ProjectionSnapshotQuery.getSnapshot:listProjects:decodeRows",
962 ),
963 ),
964 ),
965 loadThreadRows: listThreadRows(undefined).pipe(
966 Effect.mapError(
967 toPersistenceSqlOrDecodeError(
968 "ProjectionSnapshotQuery.getSnapshot:listThreads:query",
969 "ProjectionSnapshotQuery.getSnapshot:listThreads:decodeRows",
970 ),
971 ),
972 ),
973 loadMessageRows: listThreadMessageRows(undefined).pipe(
974 Effect.mapError(
975 toPersistenceSqlOrDecodeError(
976 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:query",
977 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:decodeRows",
978 ),
979 ),
980 ),
981 loadProposedPlanRows: listThreadProposedPlanRows(undefined).pipe(
982 Effect.mapError(
983 toPersistenceSqlOrDecodeError(
984 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:query",
985 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:decodeRows",
986 ),
987 ),
988 ),
989 loadActivityRows: listThreadActivityRows(undefined).pipe(
990 Effect.mapError(
991 toPersistenceSqlOrDecodeError(
992 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:query",
993 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:decodeRows",
994 ),
995 ),
996 ),
997 loadSessionRows: listThreadSessionRows(undefined).pipe(
998 Effect.mapError(
999 toPersistenceSqlOrDecodeError(
1000 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:query",
1001 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:decodeRows",
1002 ),
1003 ),
1004 ),
1005 loadCheckpointRows: listCheckpointRows(undefined).pipe(
1006 Effect.mapError(
1007 toPersistenceSqlOrDecodeError(
1008 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:query",
1009 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:decodeRows",
1010 ),
1011 ),
1012 ),
1013 loadLatestTurnRows: listLatestTurnRows(undefined).pipe(
1014 Effect.mapError(
1015 toPersistenceSqlOrDecodeError(
1016 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:query",
1017 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:decodeRows",
1018 ),
1019 ),
1020 ),
1021 });
1022
1023 const getActiveSnapshot: ProjectionSnapshotQueryShape["getActiveSnapshot"] = () =>
1024 getSnapshotFromQueries({
1025 operationPrefix: "ProjectionSnapshotQuery.getActiveSnapshot",
1026 loadProjectRows: listActiveProjectRows(undefined).pipe(
1027 Effect.mapError(
1028 toPersistenceSqlOrDecodeError(
1029 "ProjectionSnapshotQuery.getActiveSnapshot:listProjects:query",
1030 "ProjectionSnapshotQuery.getActiveSnapshot:listProjects:decodeRows",
1031 ),
1032 ),
1033 ),
1034 loadThreadRows: listActiveThreadRows(undefined).pipe(
1035 Effect.mapError(
1036 toPersistenceSqlOrDecodeError(
1037 "ProjectionSnapshotQuery.getActiveSnapshot:listThreads:query",
1038 "ProjectionSnapshotQuery.getActiveSnapshot:listThreads:decodeRows",
1039 ),
1040 ),
1041 ),
1042 loadMessageRows: listActiveThreadMessageRows(undefined).pipe(
1043 Effect.mapError(
1044 toPersistenceSqlOrDecodeError(
1045 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadMessages:query",
1046 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadMessages:decodeRows",
1047 ),
1048 ),
1049 ),
1050 loadProposedPlanRows: listActiveThreadProposedPlanRows(undefined).pipe(
1051 Effect.mapError(
1052 toPersistenceSqlOrDecodeError(
1053 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadProposedPlans:query",
1054 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadProposedPlans:decodeRows",
1055 ),
1056 ),
1057 ),
1058 loadActivityRows: listActiveThreadActivityRows(undefined).pipe(
1059 Effect.mapError(
1060 toPersistenceSqlOrDecodeError(
1061 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadActivities:query",
1062 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadActivities:decodeRows",
1063 ),
1064 ),
1065 ),
1066 loadSessionRows: listActiveThreadSessionRows(undefined).pipe(
1067 Effect.mapError(
1068 toPersistenceSqlOrDecodeError(
1069 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadSessions:query",
1070 "ProjectionSnapshotQuery.getActiveSnapshot:listThreadSessions:decodeRows",
1071 ),
1072 ),
1073 ),
1074 loadCheckpointRows: listActiveCheckpointRows(undefined).pipe(
1075 Effect.mapError(
1076 toPersistenceSqlOrDecodeError(
1077 "ProjectionSnapshotQuery.getActiveSnapshot:listCheckpoints:query",
1078 "ProjectionSnapshotQuery.getActiveSnapshot:listCheckpoints:decodeRows",
1079 ),
1080 ),
1081 ),
1082 loadLatestTurnRows: listActiveLatestTurnRows(undefined).pipe(
1083 Effect.mapError(
1084 toPersistenceSqlOrDecodeError(
1085 "ProjectionSnapshotQuery.getActiveSnapshot:listLatestTurns:query",
1086 "ProjectionSnapshotQuery.getActiveSnapshot:listLatestTurns:decodeRows",
1087 ),
1088 ),
1089 ),
1090 });
1091
710 const getCounts: ProjectionSnapshotQueryShape["getCounts"] = () =>1092 const getCounts: ProjectionSnapshotQueryShape["getCounts"] = () =>
711 readProjectionCounts(undefined).pipe(1093 readProjectionCounts(undefined).pipe(
712 Effect.mapError(1094 Effect.mapError(
@@ -723,6 +1105,16 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 10 unmodified lines
723 ),1105 ),
724 );1106 );
725 1107
1108 const listArchivedThreads: ProjectionSnapshotQueryShape["listArchivedThreads"] = () =>
1109 listArchivedThreadSummaryRows(undefined).pipe(
1110 Effect.mapError(
1111 toPersistenceSqlOrDecodeError(
1112 "ProjectionSnapshotQuery.listArchivedThreads:query",
1113 "ProjectionSnapshotQuery.listArchivedThreads:decodeRows",
1114 ),
1115 ),
1116 );
1117
726 const getActiveProjectByWorkspaceRoot: ProjectionSnapshotQueryShape["getActiveProjectByWorkspaceRoot"] =1118 const getActiveProjectByWorkspaceRoot: ProjectionSnapshotQueryShape["getActiveProjectByWorkspaceRoot"] =
727 (workspaceRoot) =>1119 (workspaceRoot) =>
728 getActiveProjectRowByWorkspaceRoot({ workspaceRoot }).pipe(1120 getActiveProjectRowByWorkspaceRoot({ workspaceRoot }).pipe(
@@ -806,7 +1198,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 77 unmodified lines
806 1198
807 return {1199 return {
808 getSnapshot,1200 getSnapshot,
1201 getActiveSnapshot,
809 getCounts,1202 getCounts,
1203 listArchivedThreads,
810 getActiveProjectByWorkspaceRoot,1204 getActiveProjectByWorkspaceRoot,
811 getFirstActiveThreadIdByProjectId,1205 getFirstActiveThreadIdByProjectId,
812 getThreadCheckpointContext,1206 getThreadCheckpointContext,
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+18−0
@@ -7,6 +7,7 @@ · 6 unmodified lines
7 * @module ProjectionSnapshotQuery7 * @module ProjectionSnapshotQuery
8 */8 */
9import type {9import type {
10 OrchestrationArchivedThreadSummary,
10 OrchestrationCheckpointSummary,11 OrchestrationCheckpointSummary,
11 OrchestrationProject,12 OrchestrationProject,
12 OrchestrationReadModel,13 OrchestrationReadModel,
@@ -44,11 +45,28 @@ export interface ProjectionSnapshotQueryShape { · 31 unmodified lines
44 */45 */
45 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel, ProjectionRepositoryError>;46 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel, ProjectionRepositoryError>;
46 47
48 /**
49 * Read the latest UI-facing orchestration snapshot, omitting archived and
50 * deleted threads plus their related child rows.
51 */
52 readonly getActiveSnapshot: () => Effect.Effect<
53 OrchestrationReadModel,
54 ProjectionRepositoryError
55 >;
56
47 /**57 /**
48 * Read aggregate projection counts without hydrating the full read model.58 * Read aggregate projection counts without hydrating the full read model.
49 */59 */
50 readonly getCounts: () => Effect.Effect<ProjectionSnapshotCounts, ProjectionRepositoryError>;60 readonly getCounts: () => Effect.Effect<ProjectionSnapshotCounts, ProjectionRepositoryError>;
51 61
62 /**
63 * Read archived thread summaries for the archive settings page.
64 */
65 readonly listArchivedThreads: () => Effect.Effect<
66 ReadonlyArray<OrchestrationArchivedThreadSummary>,
67 ProjectionRepositoryError
68 >;
69
52 /**70 /**
53 * Read the active project for an exact workspace root match.71 * Read the active project for an exact workspace root match.
54 */72 */
apps/server/src/server.test.ts+29−0
@@ -217,6 +217,8 @@ const buildAppUnderTest = (options?: { · 216 unmodified lines
217 Layer.provide(217 Layer.provide(
218 Layer.mock(ProjectionSnapshotQuery)({218 Layer.mock(ProjectionSnapshotQuery)({
219 getSnapshot: () => Effect.succeed(makeDefaultOrchestrationReadModel()),219 getSnapshot: () => Effect.succeed(makeDefaultOrchestrationReadModel()),
220 getActiveSnapshot: () => Effect.succeed(makeDefaultOrchestrationReadModel()),
221 listArchivedThreads: () => Effect.succeed([]),
220 ...options?.layers?.projectionSnapshotQuery,222 ...options?.layers?.projectionSnapshotQuery,
221 }),223 }),
222 ),224 ),
@@ -1098,11 +1100,26 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 875 unmodified lines
1098 },1100 },
1099 ],1101 ],
1100 };1102 };
1103 const archivedThreads = [
1104 {
1105 threadId: ThreadId.makeUnsafe("thread-archived"),
1106 projectId: ProjectId.makeUnsafe("project-a"),
1107 projectTitle: "Project A",
1108 workspaceRoot: "/tmp/project-a",
1109 title: "Archived Thread",
1110 worktreePath: null,
1111 createdAt: now,
1112 updatedAt: now,
1113 archivedAt: now,
1114 },
1115 ];
1101 1116
1102 yield* buildAppUnderTest({1117 yield* buildAppUnderTest({
1103 layers: {1118 layers: {
1104 projectionSnapshotQuery: {1119 projectionSnapshotQuery: {
1105 getSnapshot: () => Effect.succeed(snapshot),1120 getSnapshot: () => Effect.succeed(snapshot),
1121 getActiveSnapshot: () => Effect.succeed(snapshot),
1122 listArchivedThreads: () => Effect.succeed(archivedThreads),
1106 },1123 },
1107 orchestrationEngine: {1124 orchestrationEngine: {
1108 dispatch: () => Effect.succeed({ sequence: 7 }),1125 dispatch: () => Effect.succeed({ sequence: 7 }),
@@ -1133,6 +1150,11 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 24 unmodified lines
1133 );1150 );
1134 assert.equal(snapshotResult.snapshotSequence, 1);1151 assert.equal(snapshotResult.snapshotSequence, 1);
1135 1152
1153 const activeSnapshotResult = yield* Effect.scoped(
1154 withWsRpcClient(wsUrl, (client) => client[ORCHESTRATION_WS_METHODS.getActiveSnapshot]({})),
1155 );
1156 assert.equal(activeSnapshotResult.snapshotSequence, 1);
1157
1136 const dispatchResult = yield* Effect.scoped(1158 const dispatchResult = yield* Effect.scoped(
1137 withWsRpcClient(wsUrl, (client) =>1159 withWsRpcClient(wsUrl, (client) =>
1138 client[ORCHESTRATION_WS_METHODS.dispatchCommand]({1160 client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
@@ -1166,6 +1188,13 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 27 unmodified lines
1166 );1188 );
1167 assert.equal(fullDiffResult.diff, "full-diff");1189 assert.equal(fullDiffResult.diff, "full-diff");
1168 1190
1191 const archivedThreadsResult = yield* Effect.scoped(
1192 withWsRpcClient(wsUrl, (client) =>
1193 client[ORCHESTRATION_WS_METHODS.listArchivedThreads]({}),
1194 ),
1195 );
1196 assert.deepEqual(archivedThreadsResult, archivedThreads);
1197
1169 const replayResult = yield* Effect.scoped(1198 const replayResult = yield* Effect.scoped(
1170 withWsRpcClient(wsUrl, (client) =>1199 withWsRpcClient(wsUrl, (client) =>
1171 client[ORCHESTRATION_WS_METHODS.replayEvents]({1200 client[ORCHESTRATION_WS_METHODS.replayEvents]({
apps/server/src/serverRuntimeStartup.test.ts+2−0
@@ -61,6 +61,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa · 60 unmodified lines
61 yield* launchStartupHeartbeat.pipe(61 yield* launchStartupHeartbeat.pipe(
62 Effect.provideService(ProjectionSnapshotQuery, {62 Effect.provideService(ProjectionSnapshotQuery, {
63 getSnapshot: () => Effect.die("unused"),63 getSnapshot: () => Effect.die("unused"),
64 getActiveSnapshot: () => Effect.die("unused"),
64 getCounts: () =>65 getCounts: () =>
65 Deferred.await(releaseCounts).pipe(66 Deferred.await(releaseCounts).pipe(
66 Effect.as({67 Effect.as({
@@ -68,6 +69,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa · 1 unmodified line
68 threadCount: 3,69 threadCount: 3,
69 }),70 }),
70 ),71 ),
72 listArchivedThreads: () => Effect.succeed([]),
71 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),73 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
72 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),74 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
73 getThreadCheckpointContext: () => Effect.succeed(Option.none()),75 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
apps/server/src/ws.ts+22−0
@@ -4,7 +4,9 @@ import { · 3 unmodified lines
4 type GitManagerServiceError,4 type GitManagerServiceError,
5 OrchestrationDispatchCommandError,5 OrchestrationDispatchCommandError,
6 type OrchestrationEvent,6 type OrchestrationEvent,
7 OrchestrationGetActiveSnapshotError,
7 OrchestrationGetFullThreadDiffError,8 OrchestrationGetFullThreadDiffError,
9 OrchestrationListArchivedThreadsError,
8 OrchestrationGetSnapshotError,10 OrchestrationGetSnapshotError,
9 OrchestrationGetTurnDiffError,11 OrchestrationGetTurnDiffError,
10 ORCHESTRATION_WS_METHODS,12 ORCHESTRATION_WS_METHODS,
@@ -82,6 +84,16 @@ const WsRpcLayer = WsRpcGroup.toLayer( · 71 unmodified lines
82 }),84 }),
83 ),85 ),
84 ),86 ),
87 [ORCHESTRATION_WS_METHODS.getActiveSnapshot]: (_input) =>
88 projectionSnapshotQuery.getActiveSnapshot().pipe(
89 Effect.mapError(
90 (cause) =>
91 new OrchestrationGetActiveSnapshotError({
92 message: "Failed to load active orchestration snapshot",
93 cause,
94 }),
95 ),
96 ),
85 [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) =>97 [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) =>
86 Effect.gen(function* () {98 Effect.gen(function* () {
87 const normalizedCommand = yield* normalizeDispatchCommand(command);99 const normalizedCommand = yield* normalizeDispatchCommand(command);
@@ -116,6 +128,16 @@ const WsRpcLayer = WsRpcGroup.toLayer( · 28 unmodified lines
116 }),128 }),
117 ),129 ),
118 ),130 ),
131 [ORCHESTRATION_WS_METHODS.listArchivedThreads]: (_input) =>
132 projectionSnapshotQuery.listArchivedThreads().pipe(
133 Effect.mapError(
134 (cause) =>
135 new OrchestrationListArchivedThreadsError({
136 message: "Failed to load archived threads",
137 cause,
138 }),
139 ),
140 ),
119 [ORCHESTRATION_WS_METHODS.replayEvents]: (input) =>141 [ORCHESTRATION_WS_METHODS.replayEvents]: (input) =>
120 Stream.runCollect(142 Stream.runCollect(
121 orchestrationEngine.readEvents(143 orchestrationEngine.readEvents(
apps/web/src/components/ChatView.browser.tsx+4−1
@@ -622,7 +622,10 @@ function resolveWsRpc(body: NormalizedWsRpcRequestBody): unknown { · 621 unmodified lines
622 return customResult;622 return customResult;
623 }623 }
624 const tag = body._tag;624 const tag = body._tag;
625 if (tag === ORCHESTRATION_WS_METHODS.getSnapshot) {625 if (
626 tag === ORCHESTRATION_WS_METHODS.getSnapshot ||
627 tag === ORCHESTRATION_WS_METHODS.getActiveSnapshot
628 ) {
626 return fixture.snapshot;629 return fixture.snapshot;
627 }630 }
628 if (tag === WS_METHODS.serverGetConfig) {631 if (tag === WS_METHODS.serverGetConfig) {
apps/web/src/components/KeybindingsToast.browser.tsx+4−1
@@ -147,7 +147,10 @@ function buildFixture(): TestFixture { · 146 unmodified lines
147}147}
148 148
149function resolveWsRpc(tag: string): unknown {149function resolveWsRpc(tag: string): unknown {
150 if (tag === ORCHESTRATION_WS_METHODS.getSnapshot) {150 if (
151 tag === ORCHESTRATION_WS_METHODS.getSnapshot ||
152 tag === ORCHESTRATION_WS_METHODS.getActiveSnapshot
153 ) {
151 return fixture.snapshot;154 return fixture.snapshot;
152 }155 }
153 if (tag === WS_METHODS.serverGetConfig) {156 if (tag === WS_METHODS.serverGetConfig) {
apps/web/src/components/settings/SettingsPanels.tsx+123−71
@@ -9,7 +9,7 @@ import { · 8 unmodified lines
9 Undo2Icon,9 Undo2Icon,
10 XIcon,10 XIcon,
11} from "lucide-react";11} from "lucide-react";
12import { useQueryClient } from "@tanstack/react-query";12import { useQuery, useQueryClient } from "@tanstack/react-query";
13import { type ReactNode, useCallback, useEffect, useMemo, useRef, useState } from "react";13import { type ReactNode, useCallback, useEffect, useMemo, useRef, useState } from "react";
14import {14import {
15 PROVIDER_DISPLAY_NAMES,15 PROVIDER_DISPLAY_NAMES,
@@ -40,13 +40,13 @@ import { · 24 unmodified lines
40 setDesktopUpdateStateQueryData,40 setDesktopUpdateStateQueryData,
41 useDesktopUpdateState,41 useDesktopUpdateState,
42} from "../../lib/desktopUpdateReactQuery";42} from "../../lib/desktopUpdateReactQuery";
43import { archivedThreadsQueryOptions } from "../../lib/orchestrationReactQuery";
43import {44import {
44 MAX_CUSTOM_MODEL_LENGTH,45 MAX_CUSTOM_MODEL_LENGTH,
45 getCustomModelOptionsByProvider,46 getCustomModelOptionsByProvider,
46 resolveAppModelSelectionState,47 resolveAppModelSelectionState,
47} from "../../modelSelection";48} from "../../modelSelection";
48import { ensureNativeApi, readNativeApi } from "../../nativeApi";49import { ensureNativeApi, readNativeApi } from "../../nativeApi";
49import { useStore } from "../../store";
50import { formatRelativeTime, formatRelativeTimeLabel } from "../../timestampFormat";50import { formatRelativeTime, formatRelativeTimeLabel } from "../../timestampFormat";
51import { cn } from "../../lib/utils";51import { cn } from "../../lib/utils";
52import { Button } from "../ui/button";52import { Button } from "../ui/button";
@@ -1414,24 +1414,46 @@ export function GeneralSettingsPanel() { · 1361 unmodified lines
1414}1414}
1415 1415
1416export function ArchivedThreadsPanel() {1416export function ArchivedThreadsPanel() {
1417 const projects = useStore((store) => store.projects);1417 const {
1418 const threads = useStore((store) => store.threads);1418 data: archivedThreads = [],
1419 const { unarchiveThread, confirmAndDeleteThread } = useThreadActions();1419 error: archivedThreadsError,
1420 isPending: archivedThreadsPending,
1421 } = useQuery(archivedThreadsQueryOptions());
1422 const { unarchiveThread, deleteArchivedThread } = useThreadActions();
1420 const archivedGroups = useMemo(() => {1423 const archivedGroups = useMemo(() => {
1421 const projectById = new Map(projects.map((project) => [project.id, project] as const));1424 const groups = new Map<
1422 return [...projectById.values()]1425 string,
1423 .map((project) => ({1426 {
1424 project,1427 project: { id: string; name: string; cwd: string };
1425 threads: threads1428 threads: Array<(typeof archivedThreads)[number]>;
1426 .filter((thread) => thread.projectId === project.id && thread.archivedAt !== null)1429 }
1427 .toSorted((left, right) => {1430 >();
1428 const leftKey = left.archivedAt ?? left.createdAt;1431
1429 const rightKey = right.archivedAt ?? right.createdAt;1432 for (const thread of archivedThreads) {
1430 return rightKey.localeCompare(leftKey) || right.id.localeCompare(left.id);1433 const existing = groups.get(thread.projectId);
1431 }),1434 if (existing) {
1432 }))1435 existing.threads.push(thread);
1433 .filter((group) => group.threads.length > 0);1436 continue;
1434 }, [projects, threads]);1437 }
1438 groups.set(thread.projectId, {
1439 project: {
1440 id: thread.projectId,
1441 name: thread.projectTitle,
1442 cwd: thread.workspaceRoot,
1443 },
1444 threads: [thread],
1445 });
1446 }
1447
1448 return Array.from(groups.values(), (group) => {
1449 group.threads.sort(
1450 (left, right) =>
1451 right.archivedAt.localeCompare(left.archivedAt) ||
1452 right.threadId.localeCompare(left.threadId),
1453 );
1454 return group;
1455 });
1456 }, [archivedThreads]);
1435 1457
1436 const handleArchivedThreadContextMenu = useCallback(1458 const handleArchivedThreadContextMenu = useCallback(
1437 async (threadId: ThreadId, position: { x: number; y: number }) => {1459 async (threadId: ThreadId, position: { x: number; y: number }) => {
@@ -1459,15 +1481,43 @@ export function ArchivedThreadsPanel() { · 21 unmodified lines
1459 }1481 }
1460 1482
1461 if (clicked === "delete") {1483 if (clicked === "delete") {
1462 await confirmAndDeleteThread(threadId);1484 const thread = archivedThreads.find((entry) => entry.threadId === threadId);
1485 if (!thread) {
1486 return;
1487 }
1488 await deleteArchivedThread(thread, archivedThreads);
1463 }1489 }
1464 },1490 },
1465 [confirmAndDeleteThread, unarchiveThread],1491 [archivedThreads, deleteArchivedThread, unarchiveThread],
1466 );1492 );
1467 1493
1468 return (1494 return (
1469 <SettingsPageContainer>1495 <SettingsPageContainer>
1470 {archivedGroups.length === 0 ? (1496 {archivedThreadsError ? (
1497 <SettingsSection title="Archived threads">
1498 <Empty className="min-h-88">
1499 <EmptyHeader>
1500 <EmptyTitle>Could not load archived threads</EmptyTitle>
1501 <EmptyDescription>
1502 {archivedThreadsError instanceof Error
1503 ? archivedThreadsError.message
1504 : "An error occurred while loading archived threads."}
1505 </EmptyDescription>
1506 </EmptyHeader>
1507 </Empty>
1508 </SettingsSection>
1509 ) : null}
1510 {!archivedThreadsError && archivedThreadsPending ? (
1511 <SettingsSection title="Archived threads">
1512 <Empty className="min-h-88">
1513 <EmptyHeader>
1514 <EmptyTitle>Loading archived threads</EmptyTitle>
1515 <EmptyDescription>Fetching archived thread summaries.</EmptyDescription>
1516 </EmptyHeader>
1517 </Empty>
1518 </SettingsSection>
1519 ) : null}
1520 {!archivedThreadsError && !archivedThreadsPending && archivedGroups.length === 0 ? (
1471 <SettingsSection title="Archived threads">1521 <SettingsSection title="Archived threads">
1472 <Empty className="min-h-88">1522 <Empty className="min-h-88">
1473 <EmptyMedia variant="icon">1523 <EmptyMedia variant="icon">
@@ -1479,56 +1529,58 @@ export function ArchivedThreadsPanel() { · 5 unmodified lines
1479 </EmptyHeader>1529 </EmptyHeader>
1480 </Empty>1530 </Empty>
1481 </SettingsSection>1531 </SettingsSection>
1482 ) : (1532 ) : null}
1483 archivedGroups.map(({ project, threads: projectThreads }) => (1533 {!archivedThreadsError && !archivedThreadsPending && archivedGroups.length > 0
1484 <SettingsSection1534 ? archivedGroups.map(({ project, threads: projectThreads }) => (
1485 key={project.id}1535 <SettingsSection
1486 title={project.name}1536 key={project.id}
1487 icon={<ProjectFavicon cwd={project.cwd} />}1537 title={project.name}
1488 >1538 icon={<ProjectFavicon cwd={project.cwd} />}
1489 {projectThreads.map((thread) => (1539 >
1490 <div1540 {projectThreads.map((thread) => (
1491 key={thread.id}1541 <div
1492 className="flex items-center justify-between gap-3 border-t border-border px-4 py-3 first:border-t-0 sm:px-5"1542 key={thread.threadId}
1493 onContextMenu={(event) => {1543 className="flex items-center justify-between gap-3 border-t border-border px-4 py-3 first:border-t-0 sm:px-5"
1494 event.preventDefault();1544 onContextMenu={(event) => {
1495 void handleArchivedThreadContextMenu(thread.id, {1545 event.preventDefault();
1496 x: event.clientX,1546 void handleArchivedThreadContextMenu(thread.threadId, {
1497 y: event.clientY,1547 x: event.clientX,
1498 });1548 y: event.clientY,
1499 }}1549 });
1500 >1550 }}
1501 <div className="min-w-0 flex-1">
1502 <h3 className="truncate text-sm font-medium text-foreground">{thread.title}</h3>
1503 <p className="text-xs text-muted-foreground">
1504 Archived {formatRelativeTimeLabel(thread.archivedAt ?? thread.createdAt)}
1505 {" \u00b7 Created "}
1506 {formatRelativeTimeLabel(thread.createdAt)}
1507 </p>
1508 </div>
1509 <Button
1510 type="button"
1511 variant="outline"
1512 size="sm"
1513 className="h-7 shrink-0 cursor-pointer gap-1.5 px-2.5"
1514 onClick={() =>
1515 void unarchiveThread(thread.id).catch((error) => {
1516 toastManager.add({
1517 type: "error",
1518 title: "Failed to unarchive thread",
1519 description: error instanceof Error ? error.message : "An error occurred.",
1520 });
1521 })
1522 }
1523 >1551 >
1524 <ArchiveX className="size-3.5" />1552 <div className="min-w-0 flex-1">
1525 <span>Unarchive</span>1553 <h3 className="truncate text-sm font-medium text-foreground">{thread.title}</h3>
1526 </Button>1554 <p className="text-xs text-muted-foreground">
1527 </div>1555 Archived {formatRelativeTimeLabel(thread.archivedAt)}
1528 ))}1556 {" \u00b7 Created "}
1529 </SettingsSection>1557 {formatRelativeTimeLabel(thread.createdAt)}
1530 ))1558 </p>
1531 )}1559 </div>
1560 <Button
1561 type="button"
1562 variant="outline"
1563 size="sm"
1564 className="h-7 shrink-0 cursor-pointer gap-1.5 px-2.5"
1565 onClick={() =>
1566 void unarchiveThread(thread.threadId).catch((error) => {
1567 toastManager.add({
1568 type: "error",
1569 title: "Failed to unarchive thread",
1570 description:
1571 error instanceof Error ? error.message : "An error occurred.",
1572 });
1573 })
1574 }
1575 >
1576 <ArchiveX className="size-3.5" />
1577 <span>Unarchive</span>
1578 </Button>
1579 </div>
1580 ))}
1581 </SettingsSection>
1582 ))
1583 : null}
1532 </SettingsPageContainer>1584 </SettingsPageContainer>
1533 );1585 );
1534}1586}
apps/web/src/hooks/useThreadActions.ts+132−11
@@ -1,4 +1,4 @@
1import { ThreadId } from "@t3tools/contracts";1import { ThreadId, type OrchestrationArchivedThreadSummary } from "@t3tools/contracts";
2import { useMutation, useQueryClient } from "@tanstack/react-query";2import { useMutation, useQueryClient } from "@tanstack/react-query";
3import { useNavigate, useParams } from "@tanstack/react-router";3import { useNavigate, useParams } from "@tanstack/react-router";
4import { useCallback } from "react";4import { useCallback } from "react";
@@ -7,6 +7,7 @@ import { getFallbackThreadIdAfterDelete } from "../components/Sidebar.logic"; · 2 unmodified lines
7import { useComposerDraftStore } from "../composerDraftStore";7import { useComposerDraftStore } from "../composerDraftStore";
8import { useHandleNewThread } from "./useHandleNewThread";8import { useHandleNewThread } from "./useHandleNewThread";
9import { gitRemoveWorktreeMutationOptions } from "../lib/gitReactQuery";9import { gitRemoveWorktreeMutationOptions } from "../lib/gitReactQuery";
10import { orchestrationQueryKeys } from "../lib/orchestrationReactQuery";
10import { newCommandId } from "../lib/utils";11import { newCommandId } from "../lib/utils";
11import { readNativeApi } from "../nativeApi";12import { readNativeApi } from "../nativeApi";
12import { useStore } from "../store";13import { useStore } from "../store";
@@ -17,6 +18,7 @@ import { useSettings } from "./useSettings"; · 4 unmodified lines
17 18
18export function useThreadActions() {19export function useThreadActions() {
19 const appSettings = useSettings();20 const appSettings = useSettings();
21 const syncServerReadModel = useStore((store) => store.syncServerReadModel);
20 const clearComposerDraftForThread = useComposerDraftStore((store) => store.clearDraftThread);22 const clearComposerDraftForThread = useComposerDraftStore((store) => store.clearDraftThread);
21 const clearProjectDraftThreadById = useComposerDraftStore(23 const clearProjectDraftThreadById = useComposerDraftStore(
22 (store) => store.clearProjectDraftThreadById,24 (store) => store.clearProjectDraftThreadById,
@@ -31,6 +33,13 @@ export function useThreadActions() { · 8 unmodified lines
31 const queryClient = useQueryClient();33 const queryClient = useQueryClient();
32 const removeWorktreeMutation = useMutation(gitRemoveWorktreeMutationOptions({ queryClient }));34 const removeWorktreeMutation = useMutation(gitRemoveWorktreeMutationOptions({ queryClient }));
33 35
36 const refreshActiveSnapshot = useCallback(async () => {
37 const api = readNativeApi();
38 if (!api) return;
39 const snapshot = await api.orchestration.getActiveSnapshot();
40 syncServerReadModel(snapshot);
41 }, [syncServerReadModel]);
42
34 const archiveThread = useCallback(43 const archiveThread = useCallback(
35 async (threadId: ThreadId) => {44 async (threadId: ThreadId) => {
36 const api = readNativeApi();45 const api = readNativeApi();
@@ -46,23 +55,29 @@ export function useThreadActions() { · 9 unmodified lines
46 commandId: newCommandId(),55 commandId: newCommandId(),
47 threadId,56 threadId,
48 });57 });
58 void queryClient.invalidateQueries({ queryKey: orchestrationQueryKeys.archivedThreads() });
49 59
50 if (routeThreadId === threadId) {60 if (routeThreadId === threadId) {
51 await handleNewThread(thread.projectId);61 await handleNewThread(thread.projectId);
52 }62 }
53 },63 },
54 [handleNewThread, routeThreadId],64 [handleNewThread, queryClient, routeThreadId],
55 );65 );
56 66
57 const unarchiveThread = useCallback(async (threadId: ThreadId) => {67 const unarchiveThread = useCallback(
58 const api = readNativeApi();68 async (threadId: ThreadId) => {
59 if (!api) return;69 const api = readNativeApi();
60 await api.orchestration.dispatchCommand({70 if (!api) return;
61 type: "thread.unarchive",71 await api.orchestration.dispatchCommand({
62 commandId: newCommandId(),72 type: "thread.unarchive",
63 threadId,73 commandId: newCommandId(),
64 });74 threadId,
65 }, []);75 });
76 await refreshActiveSnapshot();
77 void queryClient.invalidateQueries({ queryKey: orchestrationQueryKeys.archivedThreads() });
78 },
79 [queryClient, refreshActiveSnapshot],
80 );
66 81
67 const deleteThread = useCallback(82 const deleteThread = useCallback(
68 async (threadId: ThreadId, opts: { deletedThreadIds?: ReadonlySet<ThreadId> } = {}) => {83 async (threadId: ThreadId, opts: { deletedThreadIds?: ReadonlySet<ThreadId> } = {}) => {
@@ -123,6 +138,7 @@ export function useThreadActions() { · 54 unmodified lines
123 commandId: newCommandId(),138 commandId: newCommandId(),
124 threadId,139 threadId,
125 });140 });
141 void queryClient.invalidateQueries({ queryKey: orchestrationQueryKeys.archivedThreads() });
126 clearComposerDraftForThread(threadId);142 clearComposerDraftForThread(threadId);
127 clearProjectDraftThreadById(thread.projectId, thread.id);143 clearProjectDraftThreadById(thread.projectId, thread.id);
128 clearTerminalState(threadId);144 clearTerminalState(threadId);
@@ -172,6 +188,110 @@ export function useThreadActions() { · 43 unmodified lines
172 navigate,188 navigate,
173 removeWorktreeMutation,189 removeWorktreeMutation,
174 routeThreadId,190 routeThreadId,
191 queryClient,
192 ],
193 );
194
195 const deleteArchivedThread = useCallback(
196 async (
197 thread: OrchestrationArchivedThreadSummary,
198 archivedThreads: ReadonlyArray<OrchestrationArchivedThreadSummary>,
199 ) => {
200 const api = readNativeApi();
201 if (!api) return;
202
203 const { projects, threads } = useStore.getState();
204 const threadProject = projects.find((project) => project.id === thread.projectId);
205
206 if (appSettings.confirmThreadDelete) {
207 const confirmed = await api.dialogs.confirm(
208 [
209 `Delete thread "${thread.title}"?`,
210 "This permanently clears conversation history for this thread.",
211 ].join("\n"),
212 );
213 if (!confirmed) {
214 return;
215 }
216 }
217
218 const normalizedWorktreePath = thread.worktreePath?.trim() || null;
219 const hasSharedWorktree =
220 normalizedWorktreePath !== null &&
221 (threads.some(
222 (entry) =>
223 entry.id !== thread.threadId &&
224 (entry.worktreePath?.trim() || null) === normalizedWorktreePath,
225 ) ||
226 archivedThreads.some(
227 (entry) =>
228 entry.threadId !== thread.threadId &&
229 (entry.worktreePath?.trim() || null) === normalizedWorktreePath,
230 ));
231 const orphanedWorktreePath = hasSharedWorktree ? null : normalizedWorktreePath;
232 const displayWorktreePath = orphanedWorktreePath
233 ? formatWorktreePathForDisplay(orphanedWorktreePath)
234 : null;
235 const canDeleteWorktree = orphanedWorktreePath !== null && threadProject !== undefined;
236 const shouldDeleteWorktree =
237 canDeleteWorktree &&
238 (await api.dialogs.confirm(
239 [
240 "This thread is the only one linked to this worktree:",
241 displayWorktreePath ?? orphanedWorktreePath,
242 "",
243 "Delete the worktree too?",
244 ].join("\n"),
245 ));
246
247 try {
248 await api.terminal.close({ threadId: thread.threadId, deleteHistory: true });
249 } catch {
250 // Terminal may already be closed.
251 }
252
253 await api.orchestration.dispatchCommand({
254 type: "thread.delete",
255 commandId: newCommandId(),
256 threadId: thread.threadId,
257 });
258 clearComposerDraftForThread(thread.threadId);
259 clearProjectDraftThreadById(thread.projectId, thread.threadId);
260 clearTerminalState(thread.threadId);
261 void queryClient.invalidateQueries({ queryKey: orchestrationQueryKeys.archivedThreads() });
262
263 if (!shouldDeleteWorktree || !orphanedWorktreePath || !threadProject) {
264 return;
265 }
266
267 try {
268 await removeWorktreeMutation.mutateAsync({
269 cwd: threadProject.cwd,
270 path: orphanedWorktreePath,
271 force: true,
272 });
273 } catch (error) {
274 const message = error instanceof Error ? error.message : "Unknown error removing worktree.";
275 console.error("Failed to remove orphaned worktree after archived thread deletion", {
276 threadId: thread.threadId,
277 projectCwd: threadProject.cwd,
278 worktreePath: orphanedWorktreePath,
279 error,
280 });
281 toastManager.add({
282 type: "error",
283 title: "Thread deleted, but worktree removal failed",
284 description: `Could not remove ${displayWorktreePath ?? orphanedWorktreePath}. ${message}`,
285 });
286 }
287 },
288 [
289 appSettings.confirmThreadDelete,
290 clearComposerDraftForThread,
291 clearProjectDraftThreadById,
292 clearTerminalState,
293 queryClient,
294 removeWorktreeMutation,
175 ],295 ],
176 );296 );
177 297
@@ -203,6 +323,7 @@ export function useThreadActions() { · 25 unmodified lines
203 archiveThread,323 archiveThread,
204 unarchiveThread,324 unarchiveThread,
205 deleteThread,325 deleteThread,
326 deleteArchivedThread,
206 confirmAndDeleteThread,327 confirmAndDeleteThread,
207 };328 };
208}329}
apps/web/src/lib/orchestrationReactQuery.ts+23−0
@@ -0,0 +1,23 @@
1import type { OrchestrationListArchivedThreadsResult } from "@t3tools/contracts";
2import { queryOptions } from "@tanstack/react-query";
3import { ensureNativeApi } from "~/nativeApi";
4
5export const orchestrationQueryKeys = {
6 all: ["orchestration"] as const,
7 archivedThreads: () => ["orchestration", "archived-threads"] as const,
8};
9
10const DEFAULT_ARCHIVED_THREADS_STALE_TIME = 15_000;
11const EMPTY_ARCHIVED_THREADS: OrchestrationListArchivedThreadsResult = [];
12
13export function archivedThreadsQueryOptions() {
14 return queryOptions({
15 queryKey: orchestrationQueryKeys.archivedThreads(),
16 queryFn: async () => {
17 const api = ensureNativeApi();
18 return api.orchestration.listArchivedThreads();
19 },
20 staleTime: DEFAULT_ARCHIVED_THREADS_STALE_TIME,
21 placeholderData: (previous) => previous ?? EMPTY_ARCHIVED_THREADS,
22 });
23}
apps/web/src/routes/__root.tsx+36−5
@@ -39,6 +39,7 @@ import { terminalRunningSubprocessFromEvent } from "../terminalActivity"; · 38 unmodified lines
39import { migrateLocalSettingsToServer } from "../hooks/useSettings";39import { migrateLocalSettingsToServer } from "../hooks/useSettings";
40import { providerQueryKeys } from "../lib/providerReactQuery";40import { providerQueryKeys } from "../lib/providerReactQuery";
41import { projectQueryKeys } from "../lib/projectReactQuery";41import { projectQueryKeys } from "../lib/projectReactQuery";
42import { orchestrationQueryKeys } from "../lib/orchestrationReactQuery";
42import { collectActiveTerminalThreadIds } from "../lib/terminalStateCleanup";43import { collectActiveTerminalThreadIds } from "../lib/terminalStateCleanup";
43import { deriveOrchestrationBatchEffects } from "../orchestrationEventEffects";44import { deriveOrchestrationBatchEffects } from "../orchestrationEventEffects";
44import { createOrchestrationRecoveryCoordinator } from "../orchestrationRecovery";45import { createOrchestrationRecoveryCoordinator } from "../orchestrationRecovery";
@@ -309,10 +310,21 @@ function EventRouter() { · 264 unmodified lines
309 },310 },
310 );311 );
311 312
312 const applyEventBatch = (events: ReadonlyArray<OrchestrationEvent>) => {313 const applyEventBatch = (
314 events: ReadonlyArray<OrchestrationEvent>,
315 ): "applied" | "snapshot-recovery-needed" => {
313 const nextEvents = recovery.markEventBatchApplied(events);316 const nextEvents = recovery.markEventBatchApplied(events);
314 if (nextEvents.length === 0) {317 if (nextEvents.length === 0) {
315 return;318 return "applied";
319 }
320
321 const knownThreadIds = new Set(useStore.getState().threads.map((thread) => thread.id));
322 const needsActiveSnapshotRecovery = nextEvents.some(
323 (event) =>
324 event.type === "thread.unarchived" && !knownThreadIds.has(event.payload.threadId),
325 );
326 if (needsActiveSnapshotRecovery) {
327 return "snapshot-recovery-needed";
316 }328 }
317 329
318 const batchEffects = deriveOrchestrationBatchEffects(nextEvents);330 const batchEffects = deriveOrchestrationBatchEffects(nextEvents);
@@ -327,6 +339,16 @@ function EventRouter() { · 8 unmodified lines
327 needsProviderInvalidation = true;339 needsProviderInvalidation = true;
328 void queryInvalidationThrottler.maybeExecute();340 void queryInvalidationThrottler.maybeExecute();
329 }341 }
342 if (
343 nextEvents.some(
344 (event) =>
345 event.type === "thread.archived" ||
346 event.type === "thread.unarchived" ||
347 event.type === "thread.deleted",
348 )
349 ) {
350 void queryClient.invalidateQueries({ queryKey: orchestrationQueryKeys.archivedThreads() });
351 }
330 352
331 applyOrchestrationEvents(nextEvents);353 applyOrchestrationEvents(nextEvents);
332 if (needsProjectUiSync) {354 if (needsProjectUiSync) {
@@ -356,6 +378,7 @@ function EventRouter() { · 23 unmodified lines
356 for (const threadId of batchEffects.removeTerminalStateThreadIds) {378 for (const threadId of batchEffects.removeTerminalStateThreadIds) {
357 removeTerminalState(threadId);379 removeTerminalState(threadId);
358 }380 }
381 return "applied";
359 };382 };
360 383
361 const recoverFromSequenceGap = async (): Promise<void> => {384 const recoverFromSequenceGap = async (): Promise<void> => {
@@ -366,7 +389,12 @@ function EventRouter() { · 4 unmodified lines
366 try {389 try {
367 const events = await api.orchestration.replayEvents(recovery.getState().latestSequence);390 const events = await api.orchestration.replayEvents(recovery.getState().latestSequence);
368 if (!disposed) {391 if (!disposed) {
369 applyEventBatch(events);392 const result = applyEventBatch(events);
393 if (result === "snapshot-recovery-needed") {
394 recovery.failReplayRecovery();
395 void fallbackToSnapshotRecovery();
396 return;
397 }
370 }398 }
371 } catch {399 } catch {
372 recovery.failReplayRecovery();400 recovery.failReplayRecovery();
@@ -385,7 +413,7 @@ function EventRouter() { · 12 unmodified lines
385 }413 }
386 414
387 try {415 try {
388 const snapshot = await api.orchestration.getSnapshot();416 const snapshot = await api.orchestration.getActiveSnapshot();
389 if (!disposed) {417 if (!disposed) {
390 syncServerReadModel(snapshot);418 syncServerReadModel(snapshot);
391 reconcileSnapshotDerivedState();419 reconcileSnapshotDerivedState();
@@ -410,7 +438,10 @@ function EventRouter() { · 18 unmodified lines
410 const unsubDomainEvent = api.orchestration.onDomainEvent((event) => {438 const unsubDomainEvent = api.orchestration.onDomainEvent((event) => {
411 const action = recovery.classifyDomainEvent(event.sequence);439 const action = recovery.classifyDomainEvent(event.sequence);
412 if (action === "apply") {440 if (action === "apply") {
413 applyEventBatch([event]);441 const result = applyEventBatch([event]);
442 if (result === "snapshot-recovery-needed") {
443 void fallbackToSnapshotRecovery();
444 }
414 return;445 return;
415 }446 }
416 if (action === "recover") {447 if (action === "recover") {
apps/web/src/wsNativeApi.test.ts+2−0
@@ -76,6 +76,8 @@ const rpcClientMock = { · 75 unmodified lines
76 },76 },
77 orchestration: {77 orchestration: {
78 getSnapshot: vi.fn(),78 getSnapshot: vi.fn(),
79 getActiveSnapshot: vi.fn(),
80 listArchivedThreads: vi.fn(),
79 dispatchCommand: vi.fn(),81 dispatchCommand: vi.fn(),
80 getTurnDiff: vi.fn(),82 getTurnDiff: vi.fn(),
81 getFullThreadDiff: vi.fn(),83 getFullThreadDiff: vi.fn(),
apps/web/src/wsNativeApi.ts+2−0
@@ -91,6 +91,8 @@ export function createWsNativeApi(): NativeApi { · 90 unmodified lines
91 },91 },
92 orchestration: {92 orchestration: {
93 getSnapshot: rpcClient.orchestration.getSnapshot,93 getSnapshot: rpcClient.orchestration.getSnapshot,
94 getActiveSnapshot: rpcClient.orchestration.getActiveSnapshot,
95 listArchivedThreads: rpcClient.orchestration.listArchivedThreads,
94 dispatchCommand: rpcClient.orchestration.dispatchCommand,96 dispatchCommand: rpcClient.orchestration.dispatchCommand,
95 getTurnDiff: rpcClient.orchestration.getTurnDiff,97 getTurnDiff: rpcClient.orchestration.getTurnDiff,
96 getFullThreadDiff: rpcClient.orchestration.getFullThreadDiff,98 getFullThreadDiff: rpcClient.orchestration.getFullThreadDiff,
apps/web/src/wsRpcClient.ts+10−0
@@ -87,9 +87,15 @@ export interface WsRpcClient { · 86 unmodified lines
87 };87 };
88 readonly orchestration: {88 readonly orchestration: {
89 readonly getSnapshot: RpcUnaryNoArgMethod<typeof ORCHESTRATION_WS_METHODS.getSnapshot>;89 readonly getSnapshot: RpcUnaryNoArgMethod<typeof ORCHESTRATION_WS_METHODS.getSnapshot>;
90 readonly getActiveSnapshot: RpcUnaryNoArgMethod<
91 typeof ORCHESTRATION_WS_METHODS.getActiveSnapshot
92 >;
90 readonly dispatchCommand: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.dispatchCommand>;93 readonly dispatchCommand: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.dispatchCommand>;
91 readonly getTurnDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getTurnDiff>;94 readonly getTurnDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getTurnDiff>;
92 readonly getFullThreadDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getFullThreadDiff>;95 readonly getFullThreadDiff: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.getFullThreadDiff>;
96 readonly listArchivedThreads: RpcUnaryNoArgMethod<
97 typeof ORCHESTRATION_WS_METHODS.listArchivedThreads
98 >;
93 readonly replayEvents: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.replayEvents>;99 readonly replayEvents: RpcUnaryMethod<typeof ORCHESTRATION_WS_METHODS.replayEvents>;
94 readonly onDomainEvent: RpcStreamMethod<typeof WS_METHODS.subscribeOrchestrationDomainEvents>;100 readonly onDomainEvent: RpcStreamMethod<typeof WS_METHODS.subscribeOrchestrationDomainEvents>;
95 };101 };
@@ -187,12 +193,16 @@ export function createWsRpcClient(transport = new WsTransport()): WsRpcClient { · 91 unmodified lines
187 orchestration: {193 orchestration: {
188 getSnapshot: () =>194 getSnapshot: () =>
189 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getSnapshot]({})),195 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getSnapshot]({})),
196 getActiveSnapshot: () =>
197 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getActiveSnapshot]({})),
190 dispatchCommand: (input) =>198 dispatchCommand: (input) =>
191 transport.request((client) => client[ORCHESTRATION_WS_METHODS.dispatchCommand](input)),199 transport.request((client) => client[ORCHESTRATION_WS_METHODS.dispatchCommand](input)),
192 getTurnDiff: (input) =>200 getTurnDiff: (input) =>
193 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getTurnDiff](input)),201 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getTurnDiff](input)),
194 getFullThreadDiff: (input) =>202 getFullThreadDiff: (input) =>
195 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getFullThreadDiff](input)),203 transport.request((client) => client[ORCHESTRATION_WS_METHODS.getFullThreadDiff](input)),
204 listArchivedThreads: () =>
205 transport.request((client) => client[ORCHESTRATION_WS_METHODS.listArchivedThreads]({})),
196 replayEvents: (input) =>206 replayEvents: (input) =>
197 transport207 transport
198 .request((client) => client[ORCHESTRATION_WS_METHODS.replayEvents](input))208 .request((client) => client[ORCHESTRATION_WS_METHODS.replayEvents](input))
packages/contracts/src/ipc.ts+3−0
@@ -44,6 +44,7 @@ import type { · 43 unmodified lines
44 OrchestrationGetFullThreadDiffResult,44 OrchestrationGetFullThreadDiffResult,
45 OrchestrationGetTurnDiffInput,45 OrchestrationGetTurnDiffInput,
46 OrchestrationGetTurnDiffResult,46 OrchestrationGetTurnDiffResult,
47 OrchestrationListArchivedThreadsResult,
47 OrchestrationEvent,48 OrchestrationEvent,
48 OrchestrationReadModel,49 OrchestrationReadModel,
49} from "./orchestration";50} from "./orchestration";
@@ -174,6 +175,8 @@ export interface NativeApi { · 124 unmodified lines
174 };175 };
175 orchestration: {176 orchestration: {
176 getSnapshot: () => Promise<OrchestrationReadModel>;177 getSnapshot: () => Promise<OrchestrationReadModel>;
178 getActiveSnapshot: () => Promise<OrchestrationReadModel>;
179 listArchivedThreads: () => Promise<OrchestrationListArchivedThreadsResult>;
177 dispatchCommand: (command: ClientOrchestrationCommand) => Promise<{ sequence: number }>;180 dispatchCommand: (command: ClientOrchestrationCommand) => Promise<{ sequence: number }>;
178 getTurnDiff: (input: OrchestrationGetTurnDiffInput) => Promise<OrchestrationGetTurnDiffResult>;181 getTurnDiff: (input: OrchestrationGetTurnDiffInput) => Promise<OrchestrationGetTurnDiffResult>;
179 getFullThreadDiff: (182 getFullThreadDiff: (
packages/contracts/src/orchestration.ts+51−0
@@ -17,9 +17,11 @@ import { · 16 unmodified lines
17 17
18export const ORCHESTRATION_WS_METHODS = {18export const ORCHESTRATION_WS_METHODS = {
19 getSnapshot: "orchestration.getSnapshot",19 getSnapshot: "orchestration.getSnapshot",
20 getActiveSnapshot: "orchestration.getActiveSnapshot",
20 dispatchCommand: "orchestration.dispatchCommand",21 dispatchCommand: "orchestration.dispatchCommand",
21 getTurnDiff: "orchestration.getTurnDiff",22 getTurnDiff: "orchestration.getTurnDiff",
22 getFullThreadDiff: "orchestration.getFullThreadDiff",23 getFullThreadDiff: "orchestration.getFullThreadDiff",
24 listArchivedThreads: "orchestration.listArchivedThreads",
23 replayEvents: "orchestration.replayEvents",25 replayEvents: "orchestration.replayEvents",
24} as const;26} as const;
25 27
@@ -985,6 +987,31 @@ export type OrchestrationGetSnapshotInput = typeof OrchestrationGetSnapshotInput · 959 unmodified lines
985const OrchestrationGetSnapshotResult = OrchestrationReadModel;987const OrchestrationGetSnapshotResult = OrchestrationReadModel;
986export type OrchestrationGetSnapshotResult = typeof OrchestrationGetSnapshotResult.Type;988export type OrchestrationGetSnapshotResult = typeof OrchestrationGetSnapshotResult.Type;
987 989
990export const OrchestrationGetActiveSnapshotInput = Schema.Struct({});
991export type OrchestrationGetActiveSnapshotInput = typeof OrchestrationGetActiveSnapshotInput.Type;
992const OrchestrationGetActiveSnapshotResult = OrchestrationReadModel;
993export type OrchestrationGetActiveSnapshotResult = typeof OrchestrationGetActiveSnapshotResult.Type;
994
995export const OrchestrationArchivedThreadSummary = Schema.Struct({
996 threadId: ThreadId,
997 projectId: ProjectId,
998 projectTitle: TrimmedNonEmptyString,
999 workspaceRoot: TrimmedNonEmptyString,
1000 title: TrimmedNonEmptyString,
1001 worktreePath: Schema.NullOr(TrimmedNonEmptyString),
1002 createdAt: IsoDateTime,
1003 updatedAt: IsoDateTime,
1004 archivedAt: IsoDateTime,
1005});
1006export type OrchestrationArchivedThreadSummary = typeof OrchestrationArchivedThreadSummary.Type;
1007
1008export const OrchestrationListArchivedThreadsInput = Schema.Struct({});
1009export type OrchestrationListArchivedThreadsInput =
1010 typeof OrchestrationListArchivedThreadsInput.Type;
1011const OrchestrationListArchivedThreadsResult = Schema.Array(OrchestrationArchivedThreadSummary);
1012export type OrchestrationListArchivedThreadsResult =
1013 typeof OrchestrationListArchivedThreadsResult.Type;
1014
988export const OrchestrationGetTurnDiffInput = TurnCountRange.mapFields(1015export const OrchestrationGetTurnDiffInput = TurnCountRange.mapFields(
989 Struct.assign({ threadId: ThreadId }),1016 Struct.assign({ threadId: ThreadId }),
990 { unsafePreserveChecks: true },1017 { unsafePreserveChecks: true },
@@ -1016,6 +1043,10 @@ export const OrchestrationRpcSchemas = { · 25 unmodified lines
1016 input: OrchestrationGetSnapshotInput,1043 input: OrchestrationGetSnapshotInput,
1017 output: OrchestrationGetSnapshotResult,1044 output: OrchestrationGetSnapshotResult,
1018 },1045 },
1046 getActiveSnapshot: {
1047 input: OrchestrationGetActiveSnapshotInput,
1048 output: OrchestrationGetActiveSnapshotResult,
1049 },
1019 dispatchCommand: {1050 dispatchCommand: {
1020 input: ClientOrchestrationCommand,1051 input: ClientOrchestrationCommand,
1021 output: DispatchResult,1052 output: DispatchResult,
@@ -1028,6 +1059,10 @@ export const OrchestrationRpcSchemas = { · 6 unmodified lines
1028 input: OrchestrationGetFullThreadDiffInput,1059 input: OrchestrationGetFullThreadDiffInput,
1029 output: OrchestrationGetFullThreadDiffResult,1060 output: OrchestrationGetFullThreadDiffResult,
1030 },1061 },
1062 listArchivedThreads: {
1063 input: OrchestrationListArchivedThreadsInput,
1064 output: OrchestrationListArchivedThreadsResult,
1065 },
1031 replayEvents: {1066 replayEvents: {
1032 input: OrchestrationReplayEventsInput,1067 input: OrchestrationReplayEventsInput,
1033 output: OrchestrationReplayEventsResult,1068 output: OrchestrationReplayEventsResult,
@@ -1042,6 +1077,14 @@ export class OrchestrationGetSnapshotError extends Schema.TaggedErrorClass<Orche · 8 unmodified lines
1042 },1077 },
1043) {}1078) {}
1044 1079
1080export class OrchestrationGetActiveSnapshotError extends Schema.TaggedErrorClass<OrchestrationGetActiveSnapshotError>()(
1081 "OrchestrationGetActiveSnapshotError",
1082 {
1083 message: TrimmedNonEmptyString,
1084 cause: Schema.optional(Schema.Defect),
1085 },
1086) {}
1087
1045export class OrchestrationDispatchCommandError extends Schema.TaggedErrorClass<OrchestrationDispatchCommandError>()(1088export class OrchestrationDispatchCommandError extends Schema.TaggedErrorClass<OrchestrationDispatchCommandError>()(
1046 "OrchestrationDispatchCommandError",1089 "OrchestrationDispatchCommandError",
1047 {1090 {
@@ -1066,6 +1109,14 @@ export class OrchestrationGetFullThreadDiffError extends Schema.TaggedErrorClass · 18 unmodified lines
1066 },1109 },
1067) {}1110) {}
1068 1111
1112export class OrchestrationListArchivedThreadsError extends Schema.TaggedErrorClass<OrchestrationListArchivedThreadsError>()(
1113 "OrchestrationListArchivedThreadsError",
1114 {
1115 message: TrimmedNonEmptyString,
1116 cause: Schema.optional(Schema.Defect),
1117 },
1118) {}
1119
1069export class OrchestrationReplayEventsError extends Schema.TaggedErrorClass<OrchestrationReplayEventsError>()(1120export class OrchestrationReplayEventsError extends Schema.TaggedErrorClass<OrchestrationReplayEventsError>()(
1070 "OrchestrationReplayEventsError",1121 "OrchestrationReplayEventsError",
1071 {1122 {
packages/contracts/src/rpc.ts+24−0
@@ -31,12 +31,16 @@ import { · 30 unmodified lines
31 OrchestrationEvent,31 OrchestrationEvent,
32 ORCHESTRATION_WS_METHODS,32 ORCHESTRATION_WS_METHODS,
33 OrchestrationDispatchCommandError,33 OrchestrationDispatchCommandError,
34 OrchestrationGetActiveSnapshotError,
35 OrchestrationGetActiveSnapshotInput,
34 OrchestrationGetFullThreadDiffError,36 OrchestrationGetFullThreadDiffError,
35 OrchestrationGetFullThreadDiffInput,37 OrchestrationGetFullThreadDiffInput,
36 OrchestrationGetSnapshotError,38 OrchestrationGetSnapshotError,
37 OrchestrationGetSnapshotInput,39 OrchestrationGetSnapshotInput,
38 OrchestrationGetTurnDiffError,40 OrchestrationGetTurnDiffError,
39 OrchestrationGetTurnDiffInput,41 OrchestrationGetTurnDiffInput,
42 OrchestrationListArchivedThreadsError,
43 OrchestrationListArchivedThreadsInput,
40 OrchestrationReplayEventsError,44 OrchestrationReplayEventsError,
41 OrchestrationReplayEventsInput,45 OrchestrationReplayEventsInput,
42 OrchestrationRpcSchemas,46 OrchestrationRpcSchemas,
@@ -263,6 +267,15 @@ export const WsOrchestrationGetSnapshotRpc = Rpc.make(ORCHESTRATION_WS_METHODS.g · 220 unmodified lines
263 error: OrchestrationGetSnapshotError,267 error: OrchestrationGetSnapshotError,
264});268});
265 269
270export const WsOrchestrationGetActiveSnapshotRpc = Rpc.make(
271 ORCHESTRATION_WS_METHODS.getActiveSnapshot,
272 {
273 payload: OrchestrationGetActiveSnapshotInput,
274 success: OrchestrationRpcSchemas.getActiveSnapshot.output,
275 error: OrchestrationGetActiveSnapshotError,
276 },
277);
278
266export const WsOrchestrationDispatchCommandRpc = Rpc.make(279export const WsOrchestrationDispatchCommandRpc = Rpc.make(
267 ORCHESTRATION_WS_METHODS.dispatchCommand,280 ORCHESTRATION_WS_METHODS.dispatchCommand,
268 {281 {
@@ -287,6 +300,15 @@ export const WsOrchestrationGetFullThreadDiffRpc = Rpc.make( · 18 unmodified lines
287 },300 },
288);301);
289 302
303export const WsOrchestrationListArchivedThreadsRpc = Rpc.make(
304 ORCHESTRATION_WS_METHODS.listArchivedThreads,
305 {
306 payload: OrchestrationListArchivedThreadsInput,
307 success: OrchestrationRpcSchemas.listArchivedThreads.output,
308 error: OrchestrationListArchivedThreadsError,
309 },
310);
311
290export const WsOrchestrationReplayEventsRpc = Rpc.make(ORCHESTRATION_WS_METHODS.replayEvents, {312export const WsOrchestrationReplayEventsRpc = Rpc.make(ORCHESTRATION_WS_METHODS.replayEvents, {
291 payload: OrchestrationReplayEventsInput,313 payload: OrchestrationReplayEventsInput,
292 success: OrchestrationRpcSchemas.replayEvents.output,314 success: OrchestrationRpcSchemas.replayEvents.output,
@@ -352,8 +374,10 @@ export const WsRpcGroup = RpcGroup.make( · 59 unmodified lines
352 WsSubscribeServerConfigRpc,374 WsSubscribeServerConfigRpc,
353 WsSubscribeServerLifecycleRpc,375 WsSubscribeServerLifecycleRpc,
354 WsOrchestrationGetSnapshotRpc,376 WsOrchestrationGetSnapshotRpc,
377 WsOrchestrationGetActiveSnapshotRpc,
355 WsOrchestrationDispatchCommandRpc,378 WsOrchestrationDispatchCommandRpc,
356 WsOrchestrationGetTurnDiffRpc,379 WsOrchestrationGetTurnDiffRpc,
357 WsOrchestrationGetFullThreadDiffRpc,380 WsOrchestrationGetFullThreadDiffRpc,
381 WsOrchestrationListArchivedThreadsRpc,
358 WsOrchestrationReplayEventsRpc,382 WsOrchestrationReplayEventsRpc,
359);383);

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t