authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-08 02:19:46-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-08 02:19:46-07:00
log1f31906c8e3003b2a9d7118c4da163544fa86586
tree3263cae17e7c9103542c2e5b7a8bc3a334b1606d
parent4d1f24ba4123279708e169fcb196f78da8f11351
signature Signed by PGP key B5690EEEBB952194

fix(server): shell snapshots no longer block other database reads while decoding (#17141)

Continues #17044 by @SunkenInTime. Co-authored-by: Dara Adedeji <daraadedeji07@gmail.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

11 files changed, 367 insertions(+), 154 deletions(-)

apps/server/integration/transferBudgetV2.integration.test.ts+2
...@@ -88,6 +88,8 @@ const layerManagement = Layer.unwrap(...@@ -88,6 +88,8 @@ const layerManagement = Layer.unwrap(
88 projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),88 projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
89 getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie),89 getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie),
90 getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie),90 getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie),
91 readShellSnapshot: (options) =>
92 projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie),
91 streamStoredEventsFrom: (input) =>93 streamStoredEventsFrom: (input) =>
92 sink.stream({ ...input, bounded: true }).pipe(Stream.orDie),94 sink.stream({ ...input, bounded: true }).pipe(Stream.orDie),
93 });95 });
apps/server/src/orchestration-v2/Orchestrator.ts+28-13
...@@ -100,6 +100,7 @@ import {...@@ -100,6 +100,7 @@ import {
100 type ProjectionRecordFilter,100 type ProjectionRecordFilter,
101 type ProjectionRecords,101 type ProjectionRecords,
102 type ProjectionCheckpointContext,102 type ProjectionCheckpointContext,
103 type ShellSnapshotOptions,
103} from "./ProjectionStore.ts";104} from "./ProjectionStore.ts";
104import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";105import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
105import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts";106import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts";
...@@ -319,11 +320,16 @@ export interface OrchestratorV2Shape {...@@ -319,11 +320,16 @@ export interface OrchestratorV2Shape {
319 },320 },
320 OrchestratorV2Error321 OrchestratorV2Error
321 >;322 >;
322 readonly getShellSnapshot: (options?: {323 readonly getShellSnapshot: (
323 readonly location?: "active" | "archive";324 options?: ShellSnapshotOptions,
324 /** Background sweeps only: skips settled threads. */325 ) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, OrchestratorV2Error>;
325 readonly unsettledOnly?: boolean;326 /** See `ProjectionStoreV2Shape.readShellSnapshot`. */
326 }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, OrchestratorV2Error>;327 readonly readShellSnapshot: (
328 options?: ShellSnapshotOptions,
329 ) => Effect.Effect<
330 Effect.Effect<OrchestrationV2ThreadShellSnapshot, OrchestratorV2Error>,
331 OrchestratorV2Error
332 >;
327 readonly getThreadShell: (333 readonly getThreadShell: (
328 threadId: ThreadId,334 threadId: ThreadId,
329 ) => Effect.Effect<OrchestrationV2ThreadShell | null, OrchestratorV2Error>;335 ) => Effect.Effect<OrchestrationV2ThreadShell | null, OrchestratorV2Error>;
...@@ -10802,6 +10808,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio...@@ -10802,6 +10808,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
10802 ),10808 ),
10803 );10809 );
1080410810
10811 const shellProjectionError = (cause: unknown) =>
10812 new OrchestratorProjectionError({ threadId: ThreadId.make("thread:shell"), cause });
10813
10805 return OrchestratorV2.of({10814 return OrchestratorV2.of({
10806 resumeQueuedRuns,10815 resumeQueuedRuns,
10807 recoverDelegatedTasks,10816 recoverDelegatedTasks,
...@@ -10845,15 +10854,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio...@@ -10845,15 +10854,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
10845 .getThreadSnapshotWindow(threadId, options)10854 .getThreadSnapshotWindow(threadId, options)
10846 .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))),10855 .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))),
10847 getShellSnapshot: (options) =>10856 getShellSnapshot: (options) =>
10848 projectionStore.getShellSnapshot(options).pipe(10857 projectionStore.getShellSnapshot(options).pipe(Effect.mapError(shellProjectionError)),
10849 Effect.mapError(10858 readShellSnapshot: (options) =>
10850 (cause) =>10859 projectionStore
10851 new OrchestratorProjectionError({10860 .readShellSnapshot(options)
10852 threadId: ThreadId.make("thread:shell"),10861 .pipe(
10853 cause,10862 Effect.mapError(shellProjectionError),
10854 }),10863 Effect.map(Effect.mapError(shellProjectionError)),
10855 ),10864 ),
10856 ),
10857 getThreadShell: (threadId) =>10865 getThreadShell: (threadId) =>
10858 projectionStore10866 projectionStore
10859 .getThreadShell(threadId)10867 .getThreadShell(threadId)
...@@ -10982,6 +10990,13 @@ const layerUnavailable: Layer.Layer<OrchestratorV2> = Layer.succeed(...@@ -10982,6 +10990,13 @@ const layerUnavailable: Layer.Layer<OrchestratorV2> = Layer.succeed(
10982 cause: "Orchestration V2 live runtime is not configured.",10990 cause: "Orchestration V2 live runtime is not configured.",
10983 }),10991 }),
10984 ),10992 ),
10993 readShellSnapshot: () =>
10994 Effect.fail(
10995 new OrchestratorProjectionError({
10996 threadId: ThreadId.make("thread:shell"),
10997 cause: "Orchestration V2 live runtime is not configured.",
10998 }),
10999 ),
10985 getThreadShell: (threadId) =>11000 getThreadShell: (threadId) =>
10986 Effect.fail(11001 Effect.fail(
10987 new OrchestratorProjectionError({11002 new OrchestratorProjectionError({
apps/server/src/orchestration-v2/ProjectionStore.test.ts+75
...@@ -9,6 +9,7 @@ import {...@@ -9,6 +9,7 @@ import {
9 MessageId,9 MessageId,
10 type ModelSelection,10 type ModelSelection,
11 type OrchestrationV2ProviderThread,11 type OrchestrationV2ProviderThread,
12 type OrchestrationV2ThreadShellSnapshot,
12 NodeId,13 NodeId,
13 ProjectId,14 ProjectId,
14 ProviderDriverKind,15 ProviderDriverKind,
...@@ -1728,6 +1729,80 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {...@@ -1728,6 +1729,80 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
1728 }),1729 }),
1729 );1730 );
17301731
1732 it.effect("reads the shell snapshot first and decodes it in a separate step", () =>
1733 Effect.gen(function* () {
1734 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1735 const sql = yield* SqlClient.SqlClient;
1736 const now = yield* DateTime.now;
1737 const createThread = (threadId: ThreadId) =>
1738 projectionStore.apply({
1739 id: EventId.make(`event:${threadId}:created`),
1740 type: "thread.created",
1741 threadId,
1742 occurredAt: now,
1743 payload: {
1744 createdBy: "user",
1745 creationSource: "web",
1746 id: threadId,
1747 projectId: ProjectId.make("project:shell-read"),
1748 title: "Shell read",
1749 providerInstanceId,
1750 modelSelection,
1751 runtimeMode: "full-access",
1752 interactionMode: "default",
1753 branch: null,
1754 worktreePath: null,
1755 activeProviderThreadId: null,
1756 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
1757 forkedFrom: null,
1758 createdAt: now,
1759 updatedAt: now,
1760 archivedAt: null,
1761 settledOverride: null,
1762 settledAt: null,
1763 lastVisitedAt: null,
1764 deletedAt: null,
1765 },
1766 });
1767 const readThreadId = ThreadId.make("thread:shell-read:before");
1768 const laterThreadId = ThreadId.make("thread:shell-read:after");
1769 const shellIds = (snapshot: OrchestrationV2ThreadShellSnapshot) =>
1770 snapshot.threads
1771 .map((thread) => thread.id)
1772 .filter((id) => id === readThreadId || id === laterThreadId);
1773
1774 yield* createThread(readThreadId);
1775 const decode = yield* sql.withTransaction(projectionStore.readShellSnapshot());
1776 // Commits between the read and the decode, as another request can while
1777 // the HTTP and WebSocket loaders decode outside their transaction.
1778 yield* createThread(laterThreadId);
1779
1780 assert.deepEqual(shellIds(yield* decode), [readThreadId]);
1781 assert.sameMembers(shellIds(yield* projectionStore.getShellSnapshot()), [
1782 readThreadId,
1783 laterThreadId,
1784 ]);
1785
1786 // No decoding under the read transaction: a payload that cannot decode
1787 // fails the returned step, not the read.
1788 const [stored] = yield* sql<{ readonly payload_json: string }>`
1789 SELECT payload_json FROM orchestration_v2_projection_threads
1790 WHERE thread_id = ${laterThreadId}
1791 `;
1792 const setPayload = (payload: string) =>
1793 sql`UPDATE orchestration_v2_projection_threads SET payload_json = ${payload}
1794 WHERE thread_id = ${laterThreadId}`;
1795 yield* setPayload("{}");
1796 const failure = yield* sql
1797 .withTransaction(projectionStore.readShellSnapshot())
1798 .pipe(
1799 Effect.flatMap(Effect.flip),
1800 Effect.ensuring(Effect.orDie(setPayload(stored!.payload_json))),
1801 );
1802 assert.strictEqual(failure._tag, "ProjectionStoreReadError");
1803 }),
1804 );
1805
1731 it.effect("does not treat visited or marked-unread state as thread activity", () =>1806 it.effect("does not treat visited or marked-unread state as thread activity", () =>
1732 Effect.gen(function* () {1807 Effect.gen(function* () {
1733 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;1808 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
apps/server/src/orchestration-v2/ProjectionStore.ts+140-99
...@@ -308,6 +308,15 @@ export interface ProjectionTimelinePage {...@@ -308,6 +308,15 @@ export interface ProjectionTimelinePage {
308 readonly hasMore: boolean;308 readonly hasMore: boolean;
309}309}
310310
311export interface ShellSnapshotOptions {
312 readonly location?: "active" | "archive";
313 /**
314 * For background sweeps, not clients: skips settled threads before any of
315 * their run, item or session rows are read.
316 */
317 readonly unsettledOnly?: boolean;
318}
319
311export interface ProjectionStoreV2Shape {320export interface ProjectionStoreV2Shape {
312 readonly getThreadAttachmentIds: (321 readonly getThreadAttachmentIds: (
313 threadId: ThreadId,322 threadId: ThreadId,
...@@ -334,14 +343,21 @@ export interface ProjectionStoreV2Shape {...@@ -334,14 +343,21 @@ export interface ProjectionStoreV2Shape {
334 readonly apply: (343 readonly apply: (
335 event: OrchestrationV2DomainEvent,344 event: OrchestrationV2DomainEvent,
336 ) => Effect.Effect<void, ProjectionStoreV2Error>;345 ) => Effect.Effect<void, ProjectionStoreV2Error>;
346 readonly getShellSnapshot: (
347 options?: ShellSnapshotOptions,
348 ) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, ProjectionStoreV2Error>;
337 /**349 /**
338 * `unsettledOnly` is for background sweeps, not clients: it skips settled350 * Runs the shell snapshot's SQL reads and returns the step that decodes them.
339 * threads before any of their run, item or session rows are read.351 * Callers that combine the shell with other reads run this inside their own
352 * transaction and the returned decode after it commits, so the shared
353 * connection is not held while thousands of rows are converted.
340 */354 */
341 readonly getShellSnapshot: (options?: {355 readonly readShellSnapshot: (
342 readonly location?: "active" | "archive";356 options?: ShellSnapshotOptions,
343 readonly unsettledOnly?: boolean;357 ) => Effect.Effect<
344 }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, ProjectionStoreV2Error>;358 Effect.Effect<OrchestrationV2ThreadShellSnapshot, ProjectionStoreV2Error>,
359 ProjectionStoreV2Error
360 >;
345 readonly getThreadShell: (361 readonly getThreadShell: (
346 threadId: ThreadId,362 threadId: ThreadId,
347 ) => Effect.Effect<OrchestrationV2ThreadShell | null, ProjectionStoreV2Error>;363 ) => Effect.Effect<OrchestrationV2ThreadShell | null, ProjectionStoreV2Error>;
...@@ -5516,103 +5532,125 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -5516,103 +5532,125 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
5516 } satisfies ShellThreadState;5532 } satisfies ShellThreadState;
5517 });5533 });
55185534
5519 const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = (options) =>5535 const shellSnapshotReadError = (cause: unknown) =>
5520 sql5536 new ProjectionStoreReadError({ threadId: ThreadId.make("thread:shell"), cause });
5521 .withTransaction(
5522 Effect.gen(function* () {
5523 const targetThreadRows = yield* selectShellThreadRows(
5524 undefined,
5525 options?.location,
5526 options?.unsettledOnly ?? false,
5527 );
5528 const targetThreadIds = new Set(
5529 targetThreadRows.map((row) => ThreadId.make(row.thread_id)),
5530 );
5531 const rowsByThreadId = new Map(
5532 targetThreadRows.map((row) => [ThreadId.make(row.thread_id), row] as const),
5533 );
5534 const pendingSourceIds = targetThreadRows.flatMap((row) =>
5535 row.forked_from_run_source_thread_id === null
5536 ? []
5537 : [ThreadId.make(row.forked_from_run_source_thread_id)],
5538 );
5539 while (pendingSourceIds.length > 0) {
5540 const sourceId = pendingSourceIds.pop();
5541 if (sourceId === undefined || rowsByThreadId.has(sourceId)) continue;
5542 const source = (yield* selectShellThreadRows(sourceId))[0];
5543 if (source === undefined) continue;
5544 rowsByThreadId.set(sourceId, source);
5545 if (source.forked_from_run_source_thread_id !== null) {
5546 pendingSourceIds.push(ThreadId.make(source.forked_from_run_source_thread_id));
5547 }
5548 }
5549 const threadRows = [...rowsByThreadId.values()];
5550 const threadIds = [...rowsByThreadId.keys()];
5551 const forkSourceIds = shellForkSourceIds(threadRows);
5552 const readForThreadIds = <A>(
5553 read: (ids: ReadonlyArray<ThreadId>) => Effect.Effect<ReadonlyArray<A>, unknown>,
5554 ids: ReadonlyArray<ThreadId> = threadIds,
5555 ) => (ids.length === 0 ? Effect.succeed([] as ReadonlyArray<A>) : read(ids));
5556 const [runRows, itemCountRows, sequenceRows, providerThreadRows, pendingTurnItemRows] =
5557 yield* Effect.all([
5558 readForThreadIds(selectShellRunRows, forkSourceIds),
5559 readForThreadIds(selectShellRunItemCounts, forkSourceIds),
5560 sql<{ readonly snapshot_sequence: number | null }>`
5561 SELECT MAX(sequence) AS snapshot_sequence
5562 FROM orchestration_events
5563 WHERE application_event_version = 2
5564 AND aggregate_kind = 'thread'
5565 `,
5566 readForThreadIds(selectShellProviderThreadRows),
5567 readForThreadIds(selectShellPendingTurnItemRows),
5568 ]);
5569
5570 const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({
5571 runRows,
5572 itemCountRows,
5573 });
5574 const { providerThreadsByThreadId, pendingTurnItemsByThreadId } =
5575 yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows });
5576 const states = yield* Effect.forEach(threadRows, (row) =>
5577 shellThreadStateFromRow({
5578 row,
5579 runOrdinalsByThreadId,
5580 itemCountsByThreadId,
5581 providerThreadsByThreadId,
5582 pendingTurnItemsByThreadId,
5583 }),
5584 );
5585 const statesByThreadId = new Map(states.map((state) => [state.thread.id, state]));
55865537
5587 const shells = states5538 const readShellSnapshotRows = (options: ShellSnapshotOptions | undefined) =>
5588 .filter((state) => targetThreadIds.has(state.thread.id))5539 Effect.gen(function* () {
5589 .map((state) =>5540 const targetThreadRows = yield* selectShellThreadRows(
5590 shellFromState({5541 undefined,
5591 state,5542 options?.location,
5592 visibleItemCount: visibleItemCountForShell({5543 options?.unsettledOnly ?? false,
5593 threadId: state.thread.id,5544 );
5594 statesByThreadId,5545 const targetThreadIds = new Set(
5595 }),5546 targetThreadRows.map((row) => ThreadId.make(row.thread_id)),
5596 }),5547 );
5597 );5548 const rowsByThreadId = new Map(
5549 targetThreadRows.map((row) => [ThreadId.make(row.thread_id), row] as const),
5550 );
5551 const pendingSourceIds = targetThreadRows.flatMap((row) =>
5552 row.forked_from_run_source_thread_id === null
5553 ? []
5554 : [ThreadId.make(row.forked_from_run_source_thread_id)],
5555 );
5556 while (pendingSourceIds.length > 0) {
5557 const sourceId = pendingSourceIds.pop();
5558 if (sourceId === undefined || rowsByThreadId.has(sourceId)) continue;
5559 const source = (yield* selectShellThreadRows(sourceId))[0];
5560 if (source === undefined) continue;
5561 rowsByThreadId.set(sourceId, source);
5562 if (source.forked_from_run_source_thread_id !== null) {
5563 pendingSourceIds.push(ThreadId.make(source.forked_from_run_source_thread_id));
5564 }
5565 }
5566 const threadRows = [...rowsByThreadId.values()];
5567 const threadIds = [...rowsByThreadId.keys()];
5568 const forkSourceIds = shellForkSourceIds(threadRows);
5569 const readForThreadIds = <A>(
5570 read: (ids: ReadonlyArray<ThreadId>) => Effect.Effect<ReadonlyArray<A>, unknown>,
5571 ids: ReadonlyArray<ThreadId> = threadIds,
5572 ) => (ids.length === 0 ? Effect.succeed([] as ReadonlyArray<A>) : read(ids));
5573 const [runRows, itemCountRows, sequenceRows, providerThreadRows, pendingTurnItemRows] =
5574 yield* Effect.all([
5575 readForThreadIds(selectShellRunRows, forkSourceIds),
5576 readForThreadIds(selectShellRunItemCounts, forkSourceIds),
5577 sql<{ readonly snapshot_sequence: number | null }>`
5578 SELECT MAX(sequence) AS snapshot_sequence
5579 FROM orchestration_events
5580 WHERE application_event_version = 2
5581 AND aggregate_kind = 'thread'
5582 `,
5583 readForThreadIds(selectShellProviderThreadRows),
5584 readForThreadIds(selectShellPendingTurnItemRows),
5585 ]);
5586 return {
5587 targetThreadIds,
5588 threadRows,
5589 runRows,
5590 itemCountRows,
5591 sequenceRows,
5592 providerThreadRows,
5593 pendingTurnItemRows,
5594 };
5595 });
55985596
5599 return {5597 const decodeShellSnapshot = ({
5600 schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,5598 targetThreadIds,
5601 snapshotSequence: sequenceRows[0]?.snapshot_sequence ?? 0,5599 threadRows,
5602 threads: shells.filter((thread) => thread.archivedAt === null),5600 runRows,
5603 archivedThreads: shells.filter((thread) => thread.archivedAt !== null),5601 itemCountRows,
5604 };5602 sequenceRows,
5603 providerThreadRows,
5604 pendingTurnItemRows,
5605 }: Effect.Success<ReturnType<typeof readShellSnapshotRows>>) =>
5606 Effect.gen(function* () {
5607 const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({
5608 runRows,
5609 itemCountRows,
5610 });
5611 const { providerThreadsByThreadId, pendingTurnItemsByThreadId } =
5612 yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows });
5613 const states = yield* Effect.forEach(threadRows, (row) =>
5614 shellThreadStateFromRow({
5615 row,
5616 runOrdinalsByThreadId,
5617 itemCountsByThreadId,
5618 providerThreadsByThreadId,
5619 pendingTurnItemsByThreadId,
5605 }),5620 }),
5606 )
5607 .pipe(
5608 Effect.mapError(
5609 (cause) =>
5610 new ProjectionStoreReadError({
5611 threadId: ThreadId.make("thread:shell"),
5612 cause,
5613 }),
5614 ),
5615 );5621 );
5622 const statesByThreadId = new Map(states.map((state) => [state.thread.id, state]));
5623
5624 const shells = states
5625 .filter((state) => targetThreadIds.has(state.thread.id))
5626 .map((state) =>
5627 shellFromState({
5628 state,
5629 visibleItemCount: visibleItemCountForShell({
5630 threadId: state.thread.id,
5631 statesByThreadId,
5632 }),
5633 }),
5634 );
5635
5636 return {
5637 schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
5638 snapshotSequence: sequenceRows[0]?.snapshot_sequence ?? 0,
5639 threads: shells.filter((thread) => thread.archivedAt === null),
5640 archivedThreads: shells.filter((thread) => thread.archivedAt !== null),
5641 };
5642 }).pipe(Effect.mapError(shellSnapshotReadError));
5643
5644 const readShellSnapshot: ProjectionStoreV2Shape["readShellSnapshot"] = (options) =>
5645 readShellSnapshotRows(options).pipe(
5646 Effect.mapError(shellSnapshotReadError),
5647 Effect.map(decodeShellSnapshot),
5648 );
5649
5650 const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = (options) =>
5651 sql
5652 .withTransaction(readShellSnapshotRows(options))
5653 .pipe(Effect.mapError(shellSnapshotReadError), Effect.flatMap(decodeShellSnapshot));
56165654
5617 // Per-thread shell for the live shell streams: reads only the target thread5655 // Per-thread shell for the live shell streams: reads only the target thread
5618 // plus its fork-source chain instead of materializing every thread. Returns5656 // plus its fork-source chain instead of materializing every thread. Returns
...@@ -5687,6 +5725,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -5687,6 +5725,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
5687 return {5725 return {
5688 apply,5726 apply,
5689 getShellSnapshot,5727 getShellSnapshot,
5728 readShellSnapshot,
5690 getThreadShell,5729 getThreadShell,
5691 getThread,5730 getThread,
5692 getSettlementCandidates,5731 getSettlementCandidates,
...@@ -5780,6 +5819,8 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(...@@ -5780,6 +5819,8 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
5780 archivedThreads: visible.filter((thread) => thread.archivedAt !== null),5819 archivedThreads: visible.filter((thread) => thread.archivedAt !== null),
5781 };5820 };
5782 }),5821 }),
5822 readShellSnapshot: (options) =>
5823 service.getShellSnapshot(options).pipe(Effect.map(Effect.succeed)),
5783 getThreadShell: (threadId) =>5824 getThreadShell: (threadId) =>
5784 Effect.gen(function* () {5825 Effect.gen(function* () {
5785 const existing = (yield* Ref.get(replayState)).projections;5826 const existing = (yield* Ref.get(replayState)).projections;
apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts+1
...@@ -213,6 +213,7 @@ it.effect(...@@ -213,6 +213,7 @@ it.effect(
213 apply: () => Effect.void,213 apply: () => Effect.void,
214 getLimitRecoveryCandidates: () => Effect.die("unused getLimitRecoveryCandidates"),214 getLimitRecoveryCandidates: () => Effect.die("unused getLimitRecoveryCandidates"),
215 getShellSnapshot: () => Effect.die("unused getShellSnapshot"),215 getShellSnapshot: () => Effect.die("unused getShellSnapshot"),
216 readShellSnapshot: () => Effect.die("unused readShellSnapshot"),
216 getThreadShell: () => Effect.die("unused getThreadShell"),217 getThreadShell: () => Effect.die("unused getThreadShell"),
217 getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),218 getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
218 getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),219 getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),
apps/server/src/orchestration-v2/ShellStream.test.ts+45
...@@ -7,9 +7,12 @@ import type {...@@ -7,9 +7,12 @@ import type {
7 OrchestrationV2ThreadShell,7 OrchestrationV2ThreadShell,
8} from "@t3tools/contracts";8} from "@t3tools/contracts";
9import { ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";9import { ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
10import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
10import * as DateTime from "effect/DateTime";11import * as DateTime from "effect/DateTime";
11import * as Duration from "effect/Duration";12import * as Duration from "effect/Duration";
12import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
14import * as Option from "effect/Option";
15import * as SqlClient from "effect/sql/SqlClient";
13import * as Stream from "effect/Stream";16import * as Stream from "effect/Stream";
14import * as TestClock from "effect/testing/TestClock";17import * as TestClock from "effect/testing/TestClock";
1518
...@@ -20,6 +23,7 @@ import {...@@ -20,6 +23,7 @@ import {
20 coalesceStoredThreadEvents,23 coalesceStoredThreadEvents,
21 composeShellStreamWithEnrichment,24 composeShellStreamWithEnrichment,
22 dedupeShellEnrichment,25 dedupeShellEnrichment,
26 loadShellSnapshotParts,
23 shellStreamItemFromEnrichmentRefresh,27 shellStreamItemFromEnrichmentRefresh,
24 shellStreamItemFromThreadShell,28 shellStreamItemFromThreadShell,
25 shellStreamItemsFromInitialSnapshot,29 shellStreamItemsFromInitialSnapshot,
...@@ -101,6 +105,47 @@ function storedThreadEvent(...@@ -101,6 +105,47 @@ function storedThreadEvent(
101 return { sequence, event: { threadId, ...event } } as OrchestrationV2StoredEvent;105 return { sequence, event: { threadId, ...event } } as OrchestrationV2StoredEvent;
102}106}
103107
108describe("loadShellSnapshotParts", () => {
109 it.effect("decodes the threads after the read transaction releases the connection", () =>
110 Effect.gen(function* () {
111 const sql = yield* SqlClient.SqlClient;
112 const steps: Array<string> = [];
113 const step = <A>(name: string, value: A) =>
114 Effect.serviceOption(sql.transactionService).pipe(
115 Effect.map((transaction) => {
116 steps.push(`${name}:${Option.isSome(transaction) ? "in" : "out of"} transaction`);
117 return value;
118 }),
119 );
120 const thread = shellFixture({});
121
122 const snapshot = yield* loadShellSnapshotParts({
123 sql,
124 readThreads: step(
125 "read threads",
126 step("decode threads", {
127 schemaVersion: 1,
128 snapshotSequence: 3,
129 threads: [thread],
130 archivedThreads: [],
131 }),
132 ),
133 listProjects: step("list projects", []),
134 latestSequence: step("latest sequence", 7),
135 });
136
137 expect(steps).toEqual([
138 "read threads:in transaction",
139 "list projects:in transaction",
140 "latest sequence:in transaction",
141 "decode threads:out of transaction",
142 ]);
143 expect(snapshot.snapshotSequence).toBe(7);
144 expect(snapshot.threads.threads).toEqual([thread]);
145 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
146 );
147});
148
104function shellFixture(overrides: Partial<OrchestrationV2ThreadShell>): OrchestrationV2ThreadShell {149function shellFixture(overrides: Partial<OrchestrationV2ThreadShell>): OrchestrationV2ThreadShell {
105 return { id: "thread-a", archivedAt: null, ...overrides } as OrchestrationV2ThreadShell;150 return { id: "thread-a", archivedAt: null, ...overrides } as OrchestrationV2ThreadShell;
106}151}
apps/server/src/orchestration-v2/ShellStream.ts+28
...@@ -15,6 +15,7 @@ import {...@@ -15,6 +15,7 @@ import {
15import * as Clock from "effect/Clock";15import * as Clock from "effect/Clock";
16import * as Effect from "effect/Effect";16import * as Effect from "effect/Effect";
17import * as Schema from "effect/Schema";17import * as Schema from "effect/Schema";
18import type * as SqlClient from "effect/sql/SqlClient";
18import * as Stream from "effect/Stream";19import * as Stream from "effect/Stream";
1920
20/** Build the regular navigation shell without duplicating the archive dataset. */21/** Build the regular navigation shell without duplicating the archive dataset. */
...@@ -32,6 +33,33 @@ export function buildActiveShellSnapshot(input: {...@@ -32,6 +33,33 @@ export function buildActiveShellSnapshot(input: {
32 };33 };
33}34}
3435
36/**
37 * Loads the thread shell, projects and sequence for the shell snapshots.
38 * One transaction covers the reads so they agree; the threads are decoded
39 * after it commits, because the server shares one SQLite connection and
40 * decoding a large shell takes longer than reading it.
41 */
42export const loadShellSnapshotParts = <E1, E2, E3, E4>(input: {
43 readonly sql: SqlClient.SqlClient;
44 readonly readThreads: Effect.Effect<Effect.Effect<OrchestrationV2ThreadShellSnapshot, E2>, E1>;
45 readonly listProjects: Effect.Effect<ReadonlyArray<OrchestrationProjectShell>, E3>;
46 readonly latestSequence: Effect.Effect<number, E4>;
47}) =>
48 Effect.gen(function* () {
49 const read = yield* input.sql.withTransaction(
50 Effect.all({
51 decodeThreads: input.readThreads,
52 projects: input.listProjects,
53 snapshotSequence: input.latestSequence,
54 }),
55 );
56 return {
57 projects: read.projects,
58 threads: yield* read.decodeThreads,
59 snapshotSequence: read.snapshotSequence,
60 };
61 });
62
35export type ShellApplicationEvent =63export type ShellApplicationEvent =
36 | Pick<64 | Pick<
37 Extract<ApplicationStoredEvent, { readonly aggregateKind: "project" }>,65 Extract<ApplicationStoredEvent, { readonly aggregateKind: "project" }>,
apps/server/src/orchestration-v2/ThreadManagementService.ts+7
...@@ -309,6 +309,12 @@ export interface ThreadManagementServiceShape {...@@ -309,6 +309,12 @@ export interface ThreadManagementServiceShape {
309 readonly getShellSnapshot: (options?: {309 readonly getShellSnapshot: (options?: {
310 readonly location?: "active" | "archive";310 readonly location?: "active" | "archive";
311 }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, Orchestrator.OrchestratorV2Error>;311 }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, Orchestrator.OrchestratorV2Error>;
312 readonly readShellSnapshot: (options?: {
313 readonly location?: "active" | "archive";
314 }) => Effect.Effect<
315 Effect.Effect<OrchestrationV2ThreadShellSnapshot, Orchestrator.OrchestratorV2Error>,
316 Orchestrator.OrchestratorV2Error
317 >;
312 readonly getThreadShell: Orchestrator.OrchestratorV2["Service"]["getThreadShell"];318 readonly getThreadShell: Orchestrator.OrchestratorV2["Service"]["getThreadShell"];
313 readonly listProjectThreads: (input: {319 readonly listProjectThreads: (input: {
314 readonly projectId: ProjectId;320 readonly projectId: ProjectId;
...@@ -818,6 +824,7 @@ const make = Effect.gen(function* () {...@@ -818,6 +824,7 @@ const make = Effect.gen(function* () {
818 getProjectThreadRecords,824 getProjectThreadRecords,
819 getProjectThread,825 getProjectThread,
820 getShellSnapshot: orchestrator.getShellSnapshot,826 getShellSnapshot: orchestrator.getShellSnapshot,
827 readShellSnapshot: orchestrator.readShellSnapshot,
821 getThreadShell: orchestrator.getThreadShell,828 getThreadShell: orchestrator.getThreadShell,
822 listProjectThreads,829 listProjectThreads,
823 sendToThread,830 sendToThread,
apps/server/src/orchestration-v2/http.ts+7-9
...@@ -31,7 +31,7 @@ import {...@@ -31,7 +31,7 @@ import {
31} from "./threadHistoryPaging.ts";31} from "./threadHistoryPaging.ts";
32import * as ThreadManagementService from "./ThreadManagementService.ts";32import * as ThreadManagementService from "./ThreadManagementService.ts";
33import * as ProjectStore from "./ProjectStore.ts";33import * as ProjectStore from "./ProjectStore.ts";
34import { buildActiveShellSnapshot } from "./ShellStream.ts";34import { buildActiveShellSnapshot, loadShellSnapshotParts } from "./ShellStream.ts";
35import { boundedSnapshotResponseFields } from "./ThreadStream.ts";35import { boundedSnapshotResponseFields } from "./ThreadStream.ts";
36import { projectThreadProjectionForWire } from "./WireProjection.ts";36import { projectThreadProjectionForWire } from "./WireProjection.ts";
3737
...@@ -95,14 +95,12 @@ export const layer = HttpApiBuilder.group(...@@ -95,14 +95,12 @@ export const layer = HttpApiBuilder.group(
95 );95 );
9696
97 const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* () {97 const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* () {
98 const base = yield* sql.withTransaction(98 const base = buildActiveShellSnapshot(
99 Effect.gen(function* () {99 yield* loadShellSnapshotParts({
100 const threads = yield* threadManagement.getShellSnapshot({ location: "active" });100 sql,
101 return buildActiveShellSnapshot({101 readThreads: threadManagement.readShellSnapshot({ location: "active" }),
102 projects: yield* projectStore.listShells(),102 listProjects: projectStore.listShells(),
103 threads,103 latestSequence: applicationEvents.latestApplicationSequence,
104 snapshotSequence: yield* applicationEvents.latestApplicationSequence,
105 });
106 }),104 }),
107 );105 );
108 const projects = yield* enrichProjectShells(base.projects);106 const projects = yield* enrichProjectShells(base.projects);
apps/server/src/relay/AgentAwarenessRelay.test.ts+1
...@@ -190,6 +190,7 @@ const makeTestRelay = Effect.fnUntraced(function* (...@@ -190,6 +190,7 @@ const makeTestRelay = Effect.fnUntraced(function* (
190 catchUp.shellSnapshotReads += 1;190 catchUp.shellSnapshotReads += 1;
191 return { schemaVersion: 2, snapshotSequence: 1, threads: [], archivedThreads: [] };191 return { schemaVersion: 2, snapshotSequence: 1, threads: [], archivedThreads: [] };
192 }),192 }),
193 readShellSnapshot: unused,
193 ensureLegacyTranscript: unused,194 ensureLegacyTranscript: unused,
194 dispatch: unused,195 dispatch: unused,
195 getTimelinePage: () => Effect.die("Unused timeline read"),196 getTimelinePage: () => Effect.die("Unused timeline read"),
apps/server/src/ws.ts+33-33
...@@ -130,6 +130,7 @@ import {...@@ -130,6 +130,7 @@ import {
130 coalesceStoredThreadEvents,130 coalesceStoredThreadEvents,
131 composeShellStreamWithEnrichment,131 composeShellStreamWithEnrichment,
132 dedupeShellEnrichment,132 dedupeShellEnrichment,
133 loadShellSnapshotParts,
133 shellStreamItemFromEnrichmentRefresh,134 shellStreamItemFromEnrichmentRefresh,
134 shellStreamItemFromThreadShell,135 shellStreamItemFromThreadShell,
135 shellStreamItemsFromInitialSnapshot,136 shellStreamItemsFromInitialSnapshot,
...@@ -967,14 +968,12 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc...@@ -967,14 +968,12 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc
967 },968 },
968 );969 );
969 const loadSnapshot = Effect.fn("ws.orchestrationV2.loadShellSnapshot")(function* () {970 const loadSnapshot = Effect.fn("ws.orchestrationV2.loadShellSnapshot")(function* () {
970 const base = yield* sql.withTransaction(971 const base = buildActiveShellSnapshot(
971 Effect.gen(function* () {972 yield* loadShellSnapshotParts({
972 const threads = yield* threadManagement.getShellSnapshot({ location: "active" });973 sql,
973 return buildActiveShellSnapshot({974 readThreads: threadManagement.readShellSnapshot({ location: "active" }),
974 projects: yield* projects.listShells(),975 listProjects: projects.listShells(),
975 threads,976 latestSequence: applicationEvents.latestApplicationSequence,
976 snapshotSequence: yield* applicationEvents.latestApplicationSequence,
977 });
978 }),977 }),
979 );978 );
980 const enriched = yield* enrichProjectShells(base.projects);979 const enriched = yield* enrichProjectShells(base.projects);
...@@ -1741,32 +1740,33 @@ const layerWsRpc = (...@@ -1741,32 +1740,33 @@ const layerWsRpc = (
1741 .refreshStatus(cwd)1740 .refreshStatus(cwd)
1742 .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid);1741 .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid);
17431742
1744 const getOrchestrationV2ArchivedShellSnapshot = sql1743 const getOrchestrationV2ArchivedShellSnapshot = Effect.gen(function* () {
1745 .withTransaction(1744 const { threads, projects, snapshotSequence } = yield* loadShellSnapshotParts({
1746 Effect.gen(function* () {1745 sql,
1747 const threads = yield* threadManagement.getShellSnapshot({ location: "archive" });1746 readThreads: threadManagement.readShellSnapshot({ location: "archive" }),
1748 return {1747 listProjects: projectStore.listShells(),
1749 schemaVersion: threads.schemaVersion,1748 latestSequence: applicationEvents.latestApplicationSequence,
1750 snapshotSequence: yield* applicationEvents.latestApplicationSequence,1749 });
1751 projects: yield* projectStore.listShells(),1750 return {
1752 threads: threads.archivedThreads,1751 schemaVersion: threads.schemaVersion,
1753 } as const;1752 snapshotSequence,
1754 }),1753 projects,
1755 )1754 threads: threads.archivedThreads,
1756 .pipe(1755 } as const;
1757 Effect.flatMap((snapshot) =>1756 }).pipe(
1758 enrichProjectShells(snapshot.projects).pipe(1757 Effect.flatMap((snapshot) =>
1759 Effect.map(({ projects }) => ({ ...snapshot, projects })),1758 enrichProjectShells(snapshot.projects).pipe(
1760 ),1759 Effect.map(({ projects }) => ({ ...snapshot, projects })),
1761 ),
1762 Effect.mapError(
1763 (cause) =>
1764 new OrchestrationV2GetShellSnapshotError({
1765 message: "Failed to load archived thread snapshot",
1766 cause,
1767 }),
1768 ),1760 ),
1769 );1761 ),
1762 Effect.mapError(
1763 (cause) =>
1764 new OrchestrationV2GetShellSnapshotError({
1765 message: "Failed to load archived thread snapshot",
1766 cause,
1767 }),
1768 ),
1769 );
17701770
1771 const subscribeOrchestrationV2ArchivedShell = Effect.fn(1771 const subscribeOrchestrationV2ArchivedShell = Effect.fn(
1772 "ws.orchestrationV2.subscribeArchivedShell",1772 "ws.orchestrationV2.subscribeArchivedShell",