authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 08:46:29-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 08:46:29-07:00
logdffb4cd3b16dc6f41aced99922950ee3083082c6
tree6d13d73e39b130a91857b12d92d5a6d12c0140f0
parent3bbbc1d9fd8b3d649c60ba0137c7dae93a6aab3f
signature Signed by PGP key B5690EEEBB952194

perf(server): use one query for buffered provider events (#9706)

10 files changed, 228 insertions(+), 65 deletions(-)

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts+5
...@@ -107,6 +107,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -107,6 +107,7 @@ describe("CheckpointDiffQuery.layer", () => {
107 toCheckpointRef,107 toCheckpointRef,
108 });108 });
109 }),109 }),
110 getThreadRuntimeContext: () => Effect.die("unused"),
110 getThreadShellById: () => Effect.succeed(Option.none()),111 getThreadShellById: () => Effect.succeed(Option.none()),
111 getThreadDetailById: () => Effect.succeed(Option.none()),112 getThreadDetailById: () => Effect.succeed(Option.none()),
112 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),113 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
...@@ -203,6 +204,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -203,6 +204,7 @@ describe("CheckpointDiffQuery.layer", () => {
203 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),204 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
204 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),205 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
205 getFullThreadDiffContext: () => Effect.die("unused"),206 getFullThreadDiffContext: () => Effect.die("unused"),
207 getThreadRuntimeContext: () => Effect.die("unused"),
206 getThreadShellById: () => Effect.succeed(Option.none()),208 getThreadShellById: () => Effect.succeed(Option.none()),
207 getThreadDetailById: () => Effect.succeed(Option.none()),209 getThreadDetailById: () => Effect.succeed(Option.none()),
208 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),210 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
...@@ -289,6 +291,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -289,6 +291,7 @@ describe("CheckpointDiffQuery.layer", () => {
289 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),291 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
290 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),292 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
291 getFullThreadDiffContext: () => Effect.die("unused"),293 getFullThreadDiffContext: () => Effect.die("unused"),
294 getThreadRuntimeContext: () => Effect.die("unused"),
292 getThreadShellById: () => Effect.succeed(Option.none()),295 getThreadShellById: () => Effect.succeed(Option.none()),
293 getThreadDetailById: () => Effect.succeed(Option.none()),296 getThreadDetailById: () => Effect.succeed(Option.none()),
294 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),297 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
...@@ -360,6 +363,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -360,6 +363,7 @@ describe("CheckpointDiffQuery.layer", () => {
360 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),363 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
361 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),364 getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
362 getFullThreadDiffContext: () => Effect.die("unused"),365 getFullThreadDiffContext: () => Effect.die("unused"),
366 getThreadRuntimeContext: () => Effect.die("unused"),
363 getThreadShellById: () => Effect.succeed(Option.none()),367 getThreadShellById: () => Effect.succeed(Option.none()),
364 getThreadDetailById: () => Effect.succeed(Option.none()),368 getThreadDetailById: () => Effect.succeed(Option.none()),
365 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),369 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
...@@ -416,6 +420,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -416,6 +420,7 @@ describe("CheckpointDiffQuery.layer", () => {
416 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),420 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
417 getThreadCheckpointContext: () => Effect.succeed(Option.none()),421 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
418 getFullThreadDiffContext: () => Effect.succeed(Option.none()),422 getFullThreadDiffContext: () => Effect.succeed(Option.none()),
423 getThreadRuntimeContext: () => Effect.die("unused"),
419 getThreadShellById: () => Effect.succeed(Option.none()),424 getThreadShellById: () => Effect.succeed(Option.none()),
420 getThreadDetailById: () => Effect.succeed(Option.none()),425 getThreadDetailById: () => Effect.succeed(Option.none()),
421 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),426 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+1
...@@ -418,6 +418,7 @@ describe("OrchestrationEngine", () => {...@@ -418,6 +418,7 @@ describe("OrchestrationEngine", () => {
418 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),418 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
419 getThreadCheckpointContext: () => Effect.succeed(Option.none()),419 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
420 getFullThreadDiffContext: () => Effect.succeed(Option.none()),420 getFullThreadDiffContext: () => Effect.succeed(Option.none()),
421 getThreadRuntimeContext: () => Effect.die("unused"),
421 getThreadShellById: () => Effect.succeed(Option.none()),422 getThreadShellById: () => Effect.succeed(Option.none()),
422 getThreadDetailById: () => Effect.succeed(Option.none()),423 getThreadDetailById: () => Effect.succeed(Option.none()),
423 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),424 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+49
...@@ -22,6 +22,7 @@ import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";...@@ -22,6 +22,7 @@ import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
22import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";22import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
23import { encodeThreadDetailPageCursor } from "../threadDetailCursor.ts";23import { encodeThreadDetailPageCursor } from "../threadDetailCursor.ts";
24import { projectThreadDetailSnapshot } from "../ActivityPayloadProjection.ts";24import { projectThreadDetailSnapshot } from "../ActivityPayloadProjection.ts";
25import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts";
2526
26const asProjectId = (value: string): ProjectId => ProjectId.make(value);27const asProjectId = (value: string): ProjectId => ProjectId.make(value);
27const asTurnId = (value: string): TurnId => TurnId.make(value);28const asTurnId = (value: string): TurnId => TurnId.make(value);
...@@ -560,6 +561,38 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {...@@ -560,6 +561,38 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
560 },561 },
561 ]);562 ]);
562 }563 }
564
565 const counter = makeSqlStatementCounter();
566 const context = yield* snapshotQuery
567 .getThreadRuntimeContext(ThreadId.make("thread-1"))
568 .pipe(Effect.withTracer(counter.tracer));
569 assert.equal(counter.count(), 1);
570 assert.equal(context._tag, "Some");
571 if (context._tag === "Some") {
572 assert.deepEqual(context.value, {
573 id: ThreadId.make("thread-1"),
574 title: "Thread 1",
575 session: snapshot.threads[0]?.session,
576 });
577 }
578
579 yield* sql`
580 UPDATE projection_thread_sessions
581 SET status = 'starting', active_turn_id = NULL, provider_name = 'claudeAgent',
582 provider_instance_id = 'claude-secondary', last_error = 'Starting another session'
583 WHERE thread_id = 'thread-1'
584 `;
585 const changedContext = yield* snapshotQuery.getThreadRuntimeContext(
586 ThreadId.make("thread-1"),
587 );
588 assert.equal(changedContext._tag, "Some");
589 if (changedContext._tag === "Some") {
590 assert.equal(changedContext.value.session?.status, "starting");
591 assert.equal(changedContext.value.session?.activeTurnId, null);
592 assert.equal(changedContext.value.session?.providerName, "claudeAgent");
593 assert.equal(changedContext.value.session?.providerInstanceId, "claude-secondary");
594 assert.equal(changedContext.value.session?.lastError, "Starting another session");
595 }
563 }),596 }),
564 );597 );
565598
...@@ -680,6 +713,22 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {...@@ -680,6 +713,22 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
680 [ThreadId.make("thread-archived")],713 [ThreadId.make("thread-archived")],
681 );714 );
682 assert.equal(archivedShellSnapshot.threads[0]?.archivedAt, "2026-04-06T00:00:06.000Z");715 assert.equal(archivedShellSnapshot.threads[0]?.archivedAt, "2026-04-06T00:00:06.000Z");
716 const activeContext = yield* snapshotQuery.getThreadRuntimeContext(
717 ThreadId.make("thread-active"),
718 );
719 assert.equal(activeContext._tag, "Some");
720 if (activeContext._tag === "Some") assert.equal(activeContext.value.session, null);
721 for (const threadId of ["thread-archived", "thread-missing"]) {
722 assert.equal(
723 (yield* snapshotQuery.getThreadRuntimeContext(ThreadId.make(threadId)))._tag,
724 "None",
725 );
726 }
727 yield* sql`UPDATE projection_threads SET deleted_at = '2026-04-06T00:00:08.000Z' WHERE thread_id = 'thread-active'`;
728 assert.equal(
729 (yield* snapshotQuery.getThreadRuntimeContext(ThreadId.make("thread-active")))._tag,
730 "None",
731 );
683 }),732 }),
684 );733 );
685734
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+57
...@@ -113,6 +113,11 @@ const ProjectionThreadActivityIdRowSchema = Schema.Struct({...@@ -113,6 +113,11 @@ const ProjectionThreadActivityIdRowSchema = Schema.Struct({
113 activityId: ProjectionThreadActivity.fields.activityId,113 activityId: ProjectionThreadActivity.fields.activityId,
114});114});
115const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;115const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;
116const ProjectionThreadRuntimeContextDbRowSchema = Schema.Struct({
117 id: ThreadId,
118 title: Schema.String,
119 session: Schema.NullOr(ProjectionThreadSessionDbRowSchema),
120});
116const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields(121const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields(
117 Struct.assign({122 Struct.assign({
118 files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)),123 files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)),
...@@ -1038,6 +1043,40 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -1038,6 +1043,40 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
1038 `,1043 `,
1039 });1044 });
10401045
1046 const getThreadRuntimeContextRow = SqlSchema.findOneOption({
1047 Request: ThreadIdLookupInput,
1048 Result: ProjectionThreadRuntimeContextDbRowSchema,
1049 execute: ({ threadId }) =>
1050 sql`
1051 SELECT
1052 threads.thread_id AS id,
1053 threads.title,
1054 sessions.thread_id AS "threadId",
1055 sessions.status,
1056 sessions.provider_name AS "providerName",
1057 sessions.provider_instance_id AS "providerInstanceId",
1058 sessions.runtime_mode AS "runtimeMode",
1059 sessions.active_turn_id AS "activeTurnId",
1060 sessions.last_error AS "lastError",
1061 sessions.updated_at AS "updatedAt"
1062 FROM projection_threads AS threads
1063 LEFT JOIN projection_thread_sessions AS sessions
1064 ON sessions.thread_id = threads.thread_id
1065 WHERE threads.thread_id = ${threadId}
1066 AND threads.deleted_at IS NULL
1067 AND threads.archived_at IS NULL
1068 LIMIT 1
1069 `.pipe(
1070 Effect.map((rows) =>
1071 rows.map((row) => ({
1072 id: row.id,
1073 title: row.title,
1074 session: row.threadId === null ? null : row,
1075 })),
1076 ),
1077 ),
1078 });
1079
1041 const listThreadMessageRowsByThread = SqlSchema.findAll({1080 const listThreadMessageRowsByThread = SqlSchema.findAll({
1042 Request: ThreadIdLookupInput,1081 Request: ThreadIdLookupInput,
1043 Result: ProjectionThreadMessageDbRowSchema,1082 Result: ProjectionThreadMessageDbRowSchema,
...@@ -2764,6 +2803,23 @@ pending_approval_requests AS (...@@ -2764,6 +2803,23 @@ pending_approval_requests AS (
2764 } satisfies OrchestrationThreadShell);2803 } satisfies OrchestrationThreadShell);
2765 });2804 });
27662805
2806 const getThreadRuntimeContext: ProjectionSnapshotQueryShape["getThreadRuntimeContext"] =
2807 Effect.fn("ProjectionSnapshotQuery.getThreadRuntimeContext")(function* (threadId) {
2808 const context = yield* getThreadRuntimeContextRow({ threadId }).pipe(
2809 Effect.mapError(
2810 toPersistenceSqlOrDecodeError(
2811 "ProjectionSnapshotQuery.getThreadRuntimeContext:query",
2812 "ProjectionSnapshotQuery.getThreadRuntimeContext:decodeRow",
2813 ),
2814 ),
2815 );
2816 return Option.map(context, (row) => ({
2817 id: row.id,
2818 title: row.title,
2819 session: row.session === null ? null : mapSessionRow(row.session),
2820 }));
2821 });
2822
2767 // Contiguous turn range bounding a windowed detail read; undefined loads the2823 // Contiguous turn range bounding a windowed detail read; undefined loads the
2768 // full thread. Resolved from a window request inside the snapshot2824 // full thread. Resolved from a window request inside the snapshot
2769 // transaction (see getThreadDetailSnapshot).2825 // transaction (see getThreadDetailSnapshot).
...@@ -3200,6 +3256,7 @@ pending_approval_requests AS (...@@ -3200,6 +3256,7 @@ pending_approval_requests AS (
3200 getThreadCheckpointContext,3256 getThreadCheckpointContext,
3201 getFullThreadDiffContext,3257 getFullThreadDiffContext,
3202 getThreadShellById,3258 getThreadShellById,
3259 getThreadRuntimeContext,
3203 getThreadDetailById,3260 getThreadDetailById,
3204 getThreadDetailSnapshot,3261 getThreadDetailSnapshot,
3205 } satisfies ProjectionSnapshotQueryShape;3262 } satisfies ProjectionSnapshotQueryShape;
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+97-62
...@@ -24,6 +24,7 @@ import {...@@ -24,6 +24,7 @@ import {
24 TurnId,24 TurnId,
25} from "@t3tools/contracts";25} from "@t3tools/contracts";
26import * as Clock from "effect/Clock";26import * as Clock from "effect/Clock";
27import * as Deferred from "effect/Deferred";
27import * as Effect from "effect/Effect";28import * as Effect from "effect/Effect";
28import * as Exit from "effect/Exit";29import * as Exit from "effect/Exit";
29import * as Layer from "effect/Layer";30import * as Layer from "effect/Layer";
...@@ -31,6 +32,7 @@ import * as ManagedRuntime from "effect/ManagedRuntime";...@@ -31,6 +32,7 @@ import * as ManagedRuntime from "effect/ManagedRuntime";
31import * as PubSub from "effect/PubSub";32import * as PubSub from "effect/PubSub";
32import * as Scope from "effect/Scope";33import * as Scope from "effect/Scope";
33import * as Stream from "effect/Stream";34import * as Stream from "effect/Stream";
35import * as Tracer from "effect/Tracer";
34import { it as effectIt } from "@effect/vitest";36import { it as effectIt } from "@effect/vitest";
35import { afterEach, describe, expect, it } from "vite-plus/test";37import { afterEach, describe, expect, it } from "vite-plus/test";
3638
...@@ -55,6 +57,7 @@ import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"...@@ -55,6 +57,7 @@ import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"
55import { ServerConfig } from "../../config.ts";57import { ServerConfig } from "../../config.ts";
56import { ServerSettingsService } from "../../serverSettings.ts";58import { ServerSettingsService } from "../../serverSettings.ts";
57import * as NodeServices from "@effect/platform-node/NodeServices";59import * as NodeServices from "@effect/platform-node/NodeServices";
60import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts";
5861
59function makeTestServerSettingsLayer(overrides: Partial<ServerSettings> = {}) {62function makeTestServerSettingsLayer(overrides: Partial<ServerSettings> = {}) {
60 return ServerSettingsService.layerTest(overrides);63 return ServerSettingsService.layerTest(overrides);
...@@ -98,7 +101,12 @@ function isLegacyTurnCompletedEvent(...@@ -98,7 +101,12 @@ function isLegacyTurnCompletedEvent(
98}101}
99102
100function createProviderServiceHarness() {103function createProviderServiceHarness() {
101 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());104 const runtimeEventPubSub = Effect.runSync(
105 PubSub.unbounded<{
106 readonly events: ReadonlyArray<ProviderRuntimeEvent>;
107 readonly enqueued?: Deferred.Deferred<void>;
108 }>(),
109 );
102 const runtimeSessions: ProviderSession[] = [];110 const runtimeSessions: ProviderSession[] = [];
103111
104 const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;112 const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;
...@@ -129,7 +137,16 @@ function createProviderServiceHarness() {...@@ -129,7 +137,16 @@ function createProviderServiceHarness() {
129 rollbackConversation: () => unsupported(),137 rollbackConversation: () => unsupported(),
130 uploadFeedback: () => unsupported(),138 uploadFeedback: () => unsupported(),
131 get streamEvents() {139 get streamEvents() {
132 return Stream.fromPubSub(runtimeEventPubSub);140 return Stream.fromPubSub(runtimeEventPubSub).pipe(
141 Stream.flatMap(({ events, enqueued }) =>
142 Stream.concat(
143 Stream.fromIterable(events),
144 enqueued
145 ? Stream.fromEffect(Deferred.succeed(enqueued, undefined)).pipe(Stream.drain)
146 : Stream.empty,
147 ),
148 ),
149 );
133 },150 },
134 };151 };
135152
...@@ -158,12 +175,24 @@ function createProviderServiceHarness() {...@@ -158,12 +175,24 @@ function createProviderServiceHarness() {
158 };175 };
159176
160 const emit = (event: LegacyProviderRuntimeEvent): void => {177 const emit = (event: LegacyProviderRuntimeEvent): void => {
161 Effect.runSync(PubSub.publish(runtimeEventPubSub, normalizeLegacyEvent(event)));178 Effect.runSync(PubSub.publish(runtimeEventPubSub, { events: [normalizeLegacyEvent(event)] }));
162 };179 };
163180
181 const emitAndWaitForEnqueue = Effect.fnUntraced(function* (
182 events: ReadonlyArray<LegacyProviderRuntimeEvent>,
183 ) {
184 const enqueued = yield* Deferred.make<void>();
185 yield* PubSub.publish(runtimeEventPubSub, {
186 events: events.map(normalizeLegacyEvent),
187 enqueued,
188 });
189 yield* Deferred.await(enqueued);
190 });
191
164 return {192 return {
165 service,193 service,
166 emit,194 emit,
195 emitAndWaitForEnqueue,
167 setSession,196 setSession,
168 };197 };
169}198}
...@@ -232,6 +261,7 @@ describe("ProviderRuntimeIngestion", () => {...@@ -232,6 +261,7 @@ describe("ProviderRuntimeIngestion", () => {
232 const workspaceRoot = makeTempDir("t3-provider-project-");261 const workspaceRoot = makeTempDir("t3-provider-project-");
233 NodeFS.mkdirSync(NodePath.join(workspaceRoot, ".git"));262 NodeFS.mkdirSync(NodePath.join(workspaceRoot, ".git"));
234 const provider = createProviderServiceHarness();263 const provider = createProviderServiceHarness();
264 const sqlCounter = makeSqlStatementCounter();
235 const orchestrationLayer = OrchestrationEngineLive.pipe(265 const orchestrationLayer = OrchestrationEngineLive.pipe(
236 Layer.provide(OrchestrationProjectionSnapshotQueryLive),266 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
237 Layer.provide(OrchestrationProjectionPipelineLive),267 Layer.provide(OrchestrationProjectionPipelineLive),
...@@ -256,15 +286,22 @@ describe("ProviderRuntimeIngestion", () => {...@@ -256,15 +286,22 @@ describe("ProviderRuntimeIngestion", () => {
256 Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)),286 Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)),
257 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),287 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
258 Layer.provideMerge(NodeServices.layer),288 Layer.provideMerge(NodeServices.layer),
289 Layer.provideMerge(Layer.succeed(Tracer.Tracer, sqlCounter.tracer)),
259 );290 );
260 runtime = ManagedRuntime.make(layer);291 const testRuntime = ManagedRuntime.make(layer);
261 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));292 runtime = testRuntime;
262 const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery));293 const engine = await testRuntime.runPromise(Effect.service(OrchestrationEngineService));
263 const ingestion = await runtime.runPromise(Effect.service(ProviderRuntimeIngestionService));294 const snapshotQuery = await testRuntime.runPromise(Effect.service(ProjectionSnapshotQuery));
295 const ingestion = await testRuntime.runPromise(Effect.service(ProviderRuntimeIngestionService));
264 scope = await Effect.runPromise(Scope.make("sequential"));296 scope = await Effect.runPromise(Scope.make("sequential"));
265 await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope)));297 await testRuntime.runPromise(ingestion.start().pipe(Scope.provide(scope)));
266 const drain = () => Effect.runPromise(ingestion.drain);298 const drain = () => testRuntime.runPromise(ingestion.drain);
267 const dispatch = (command: OrchestrationCommand) => Effect.runPromise(engine.dispatch(command));299 const dispatch = (command: OrchestrationCommand) =>
300 testRuntime.runPromise(engine.dispatch(command));
301 const emitAndDrain = (events: ReadonlyArray<LegacyProviderRuntimeEvent>) =>
302 testRuntime.runPromise(
303 provider.emitAndWaitForEnqueue(events).pipe(Effect.andThen(ingestion.drain)),
304 );
268305
269 const createdAt = "2026-01-01T00:00:00.000Z";306 const createdAt = "2026-01-01T00:00:00.000Z";
270 await dispatch({307 await dispatch({
...@@ -322,8 +359,10 @@ describe("ProviderRuntimeIngestion", () => {...@@ -322,8 +359,10 @@ describe("ProviderRuntimeIngestion", () => {
322 return {359 return {
323 engine,360 engine,
324 dispatch,361 dispatch,
325 readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),362 readModel: () => testRuntime.runPromise(snapshotQuery.getSnapshot()),
326 emit: provider.emit,363 emit: provider.emit,
364 emitAndDrain,
365 sqlCount: sqlCounter.count,
327 setProviderSession: provider.setSession,366 setProviderSession: provider.setSession,
328 drain,367 drain,
329 };368 };
...@@ -2104,39 +2143,40 @@ describe("ProviderRuntimeIngestion", () => {...@@ -2104,39 +2143,40 @@ describe("ProviderRuntimeIngestion", () => {
2104 expect(proposedPlan?.planMarkdown).toBe("## Buffered plan\n\n- first\n- second");2143 expect(proposedPlan?.planMarkdown).toBe("## Buffered plan\n\n- first\n- second");
2105 });2144 });
21062145
2107 it("buffers assistant deltas by default until completion", async () => {2146 it("buffers assistant deltas with one lifecycle query per event until completion", async () => {
2108 const harness = await createHarness();2147 const harness = await createHarness();
2109 const now = "2026-01-01T00:00:00.000Z";2148 const now = "2026-01-01T00:00:00.000Z";
21102149
2111 harness.emit({2150 await harness.emitAndDrain([
2112 type: "turn.started",2151 {
2113 eventId: asEventId("evt-turn-started-buffered"),2152 type: "turn.started",
2114 provider: ProviderDriverKind.make("codex"),2153 eventId: asEventId("evt-turn-started-buffered"),
2115 createdAt: now,2154 provider: ProviderDriverKind.make("codex"),
2116 threadId: asThreadId("thread-1"),2155 createdAt: now,
2117 turnId: asTurnId("turn-buffered"),2156 threadId: asThreadId("thread-1"),
2118 });2157 turnId: asTurnId("turn-buffered"),
2119 await waitForThread(
2120 harness.readModel,
2121 (thread) =>
2122 thread.session?.status === "running" && thread.session?.activeTurnId === "turn-buffered",
2123 );
2124
2125 harness.emit({
2126 type: "content.delta",
2127 eventId: asEventId("evt-message-delta-buffered"),
2128 provider: ProviderDriverKind.make("codex"),
2129 createdAt: now,
2130 threadId: asThreadId("thread-1"),
2131 turnId: asTurnId("turn-buffered"),
2132 itemId: asItemId("item-buffered"),
2133 payload: {
2134 streamKind: "assistant_text",
2135 delta: "buffer me",
2136 },2158 },
2137 });2159 ]);
2160
2161 const eventCount = 1_000;
2162 const before = harness.sqlCount();
2163 await harness.emitAndDrain(
2164 Array.from({ length: eventCount }, (_, index) => ({
2165 type: "content.delta",
2166 eventId: asEventId(`evt-message-delta-buffered-${index}`),
2167 provider: ProviderDriverKind.make("codex"),
2168 createdAt: now,
2169 threadId: asThreadId("thread-1"),
2170 turnId: asTurnId("turn-buffered"),
2171 itemId: asItemId("item-buffered"),
2172 payload: {
2173 streamKind: "assistant_text",
2174 delta: "a",
2175 },
2176 })),
2177 );
2178 expect(harness.sqlCount() - before).toBe(eventCount);
21382179
2139 await harness.drain();
2140 const midReadModel = await harness.readModel();2180 const midReadModel = await harness.readModel();
2141 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));2181 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
2142 expect(2182 expect(
...@@ -2145,30 +2185,25 @@ describe("ProviderRuntimeIngestion", () => {...@@ -2145,30 +2185,25 @@ describe("ProviderRuntimeIngestion", () => {
2145 ),2185 ),
2146 ).toBe(false);2186 ).toBe(false);
21472187
2148 harness.emit({2188 await harness.emitAndDrain([
2149 type: "item.completed",2189 {
2150 eventId: asEventId("evt-message-completed-buffered"),2190 type: "item.completed",
2151 provider: ProviderDriverKind.make("codex"),2191 eventId: asEventId("evt-message-completed-buffered"),
2152 createdAt: now,2192 provider: ProviderDriverKind.make("codex"),
2153 threadId: asThreadId("thread-1"),2193 createdAt: now,
2154 turnId: asTurnId("turn-buffered"),2194 threadId: asThreadId("thread-1"),
2155 itemId: asItemId("item-buffered"),2195 turnId: asTurnId("turn-buffered"),
2156 payload: {2196 itemId: asItemId("item-buffered"),
2157 itemType: "assistant_message",2197 payload: {
2158 status: "completed",2198 itemType: "assistant_message",
2199 status: "completed",
2200 },
2159 },2201 },
2160 });2202 ]);
21612203 const readModel = await harness.readModel();
2162 const thread = await waitForThread(harness.readModel, (entry) =>2204 const thread = readModel.threads.find((entry) => entry.id === asThreadId("thread-1"));
2163 entry.messages.some(2205 const message = thread?.messages.find((entry) => entry.id === "assistant:item-buffered");
2164 (message: ProviderRuntimeTestMessage) =>2206 expect(message?.text).toBe("a".repeat(eventCount));
2165 message.id === "assistant:item-buffered" && !message.streaming,
2166 ),
2167 );
2168 const message = thread.messages.find(
2169 (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-buffered",
2170 );
2171 expect(message?.text).toBe("buffer me");
2172 expect(message?.streaming).toBe(false);2207 expect(message?.streaming).toBe(false);
2173 });2208 });
21742209
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+5-3
...@@ -1018,9 +1018,11 @@ const make = Effect.gen(function* () {...@@ -1018,9 +1018,11 @@ const make = Effect.gen(function* () {
1018 .pipe(Effect.map(Option.getOrUndefined));1018 .pipe(Effect.map(Option.getOrUndefined));
1019 });1019 });
10201020
1021 const resolveThreadShell = Effect.fn("resolveThreadShell")(function* (threadId: ThreadId) {1021 const resolveThreadRuntimeContext = Effect.fn("resolveThreadRuntimeContext")(function* (
1022 threadId: ThreadId,
1023 ) {
1022 return yield* projectionSnapshotQuery1024 return yield* projectionSnapshotQuery
1023 .getThreadShellById(threadId)1025 .getThreadRuntimeContext(threadId)
1024 .pipe(Effect.map(Option.getOrUndefined));1026 .pipe(Effect.map(Option.getOrUndefined));
1025 });1027 });
10261028
...@@ -1562,7 +1564,7 @@ const make = Effect.gen(function* () {...@@ -1562,7 +1564,7 @@ const make = Effect.gen(function* () {
1562 return;1564 return;
1563 }1565 }
15641566
1565 const thread = yield* resolveThreadShell(event.threadId);1567 const thread = yield* resolveThreadRuntimeContext(event.threadId);
1566 if (!thread) return;1568 if (!thread) return;
15671569
1568 let loadedThreadDetail: OrchestrationThread | null | undefined;1570 let loadedThreadDetail: OrchestrationThread | null | undefined;
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+8
...@@ -193,6 +193,14 @@ export interface ProjectionSnapshotQueryShape {...@@ -193,6 +193,14 @@ export interface ProjectionSnapshotQueryShape {
193 threadId: ThreadId,193 threadId: ThreadId,
194 ) => Effect.Effect<Option.Option<OrchestrationThreadShell>, ProjectionRepositoryError>;194 ) => Effect.Effect<Option.Option<OrchestrationThreadShell>, ProjectionRepositoryError>;
195195
196 /** Read the active thread and session facts used to ingest provider events. */
197 readonly getThreadRuntimeContext: (
198 threadId: ThreadId,
199 ) => Effect.Effect<
200 Option.Option<Pick<OrchestrationThreadShell, "id" | "title" | "session">>,
201 ProjectionRepositoryError
202 >;
203
196 /**204 /**
197 * Read a single active thread detail snapshot by id.205 * Read a single active thread detail snapshot by id.
198 */206 */
apps/server/src/project/ProjectSetupScriptRunner.test.ts+1
...@@ -43,6 +43,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>...@@ -43,6 +43,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
43 getFirstActiveThreadIdByProjectId: () => Effect.die("unused"),43 getFirstActiveThreadIdByProjectId: () => Effect.die("unused"),
44 getThreadCheckpointContext: () => Effect.die("unused"),44 getThreadCheckpointContext: () => Effect.die("unused"),
45 getFullThreadDiffContext: () => Effect.die("unused"),45 getFullThreadDiffContext: () => Effect.die("unused"),
46 getThreadRuntimeContext: () => Effect.die("unused"),
46 getThreadShellById: () => Effect.die("unused"),47 getThreadShellById: () => Effect.die("unused"),
47 getThreadDetailById: () => Effect.die("unused"),48 getThreadDetailById: () => Effect.die("unused"),
48 getThreadDetailSnapshot: () => Effect.die("unused"),49 getThreadDetailSnapshot: () => Effect.die("unused"),
apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+1
...@@ -219,6 +219,7 @@ describe("ProviderSessionReaper", () => {...@@ -219,6 +219,7 @@ describe("ProviderSessionReaper", () => {
219 getFirstActiveThreadIdByProjectId: () => Effect.die("unused"),219 getFirstActiveThreadIdByProjectId: () => Effect.die("unused"),
220 getThreadCheckpointContext: () => Effect.die("unused"),220 getThreadCheckpointContext: () => Effect.die("unused"),
221 getFullThreadDiffContext: () => Effect.die("unused"),221 getFullThreadDiffContext: () => Effect.die("unused"),
222 getThreadRuntimeContext: () => Effect.die("unused"),
222 getThreadShellById: (threadId) =>223 getThreadShellById: (threadId) =>
223 Effect.succeed(224 Effect.succeed(
224 input.readModel.threads.find((thread) => thread.id === threadId)225 input.readModel.threads.find((thread) => thread.id === threadId)
apps/server/src/serverRuntimeStartup.test.ts+4
...@@ -138,6 +138,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa...@@ -138,6 +138,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa
138 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),138 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
139 getThreadCheckpointContext: () => Effect.succeed(Option.none()),139 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
140 getFullThreadDiffContext: () => Effect.succeed(Option.none()),140 getFullThreadDiffContext: () => Effect.succeed(Option.none()),
141 getThreadRuntimeContext: () => Effect.die("unused"),
141 getThreadShellById: () => Effect.succeed(Option.none()),142 getThreadShellById: () => Effect.succeed(Option.none()),
142 getThreadDetailById: () => Effect.succeed(Option.none()),143 getThreadDetailById: () => Effect.succeed(Option.none()),
143 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),144 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
...@@ -211,6 +212,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa...@@ -211,6 +212,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
211 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.some(bootstrapThreadId)),212 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.some(bootstrapThreadId)),
212 getThreadCheckpointContext: () => Effect.succeed(Option.none()),213 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
213 getFullThreadDiffContext: () => Effect.succeed(Option.none()),214 getFullThreadDiffContext: () => Effect.succeed(Option.none()),
215 getThreadRuntimeContext: () => Effect.die("unused"),
214 getThreadShellById: () => Effect.die("unused"),216 getThreadShellById: () => Effect.die("unused"),
215 getThreadDetailById: () => Effect.die("unused"),217 getThreadDetailById: () => Effect.die("unused"),
216 getThreadDetailSnapshot: () => Effect.die("unused"),218 getThreadDetailSnapshot: () => Effect.die("unused"),
...@@ -265,6 +267,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when...@@ -265,6 +267,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when
265 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),267 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
266 getThreadCheckpointContext: () => Effect.succeed(Option.none()),268 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
267 getFullThreadDiffContext: () => Effect.succeed(Option.none()),269 getFullThreadDiffContext: () => Effect.succeed(Option.none()),
270 getThreadRuntimeContext: () => Effect.die("unused"),
268 getThreadShellById: () => Effect.die("unused"),271 getThreadShellById: () => Effect.die("unused"),
269 getThreadDetailById: () => Effect.die("unused"),272 getThreadDetailById: () => Effect.die("unused"),
270 getThreadDetailSnapshot: () => Effect.die("unused"),273 getThreadDetailSnapshot: () => Effect.die("unused"),
...@@ -328,6 +331,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa...@@ -328,6 +331,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
328 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),331 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
329 getThreadCheckpointContext: () => Effect.succeed(Option.none()),332 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
330 getFullThreadDiffContext: () => Effect.succeed(Option.none()),333 getFullThreadDiffContext: () => Effect.succeed(Option.none()),
334 getThreadRuntimeContext: () => Effect.die("unused"),
331 getThreadShellById: () => Effect.die("unused"),335 getThreadShellById: () => Effect.die("unused"),
332 getThreadDetailById: () => Effect.die("unused"),336 getThreadDetailById: () => Effect.die("unused"),
333 getThreadDetailSnapshot: () => Effect.die("unused"),337 getThreadDetailSnapshot: () => Effect.die("unused"),