authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 12:12:49-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 12:12:49-07:00
log6365919f2e5bcfb4fa4020b95e19af26ae40979f
treea0a61525de2cc0b0aa99ae9a091ee2ff20479133
parent77b655c47b82c9b2aad09e447117b7618a16e21d
signature Signed by PGP key B5690EEEBB952194

perf(server): skip history reads for metadata commands (#9758)

2 files changed, 69 insertions(+), 43 deletions(-)

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+46-26
...@@ -29,6 +29,7 @@ import * as Deferred from "effect/Deferred";...@@ -29,6 +29,7 @@ import * as Deferred from "effect/Deferred";
29import * as Exit from "effect/Exit";29import * as Exit from "effect/Exit";
30import * as Layer from "effect/Layer";30import * as Layer from "effect/Layer";
31import * as ManagedRuntime from "effect/ManagedRuntime";31import * as ManagedRuntime from "effect/ManagedRuntime";
32import * as Option from "effect/Option";
32import * as PubSub from "effect/PubSub";33import * as PubSub from "effect/PubSub";
33import * as Scope from "effect/Scope";34import * as Scope from "effect/Scope";
34import * as Stream from "effect/Stream";35import * as Stream from "effect/Stream";
...@@ -170,6 +171,7 @@ describe("ProviderCommandReactor", () => {...@@ -170,6 +171,7 @@ describe("ProviderCommandReactor", () => {
170 readonly threadModelSelection?: ModelSelection;171 readonly threadModelSelection?: ModelSelection;
171 readonly sessionModelSwitch?: "unsupported" | "in-session";172 readonly sessionModelSwitch?: "unsupported" | "in-session";
172 readonly requiresNewThreadForModelChange?: boolean;173 readonly requiresNewThreadForModelChange?: boolean;
174 readonly unreadableHistory?: boolean;
173 readonly titleRegenerationCompletionDispatchFailures?: number;175 readonly titleRegenerationCompletionDispatchFailures?: number;
174 readonly titleRegenerationBeforeStart?: "one" | "two";176 readonly titleRegenerationBeforeStart?: "one" | "two";
175 readonly serverActivation?: Effect.Effect<void>;177 readonly serverActivation?: Effect.Effect<void>;
...@@ -512,6 +514,23 @@ describe("ProviderCommandReactor", () => {...@@ -512,6 +514,23 @@ describe("ProviderCommandReactor", () => {
512 createdAt: now,514 createdAt: now,
513 }),515 }),
514 );516 );
517 if (input?.unreadableHistory === true) {
518 // Metadata commands must not decode this unrelated message body.
519 await runtime.runPromise(
520 Effect.gen(function* () {
521 const sql = yield* SqlClient.SqlClient;
522 yield* sql`
523 INSERT INTO projection_thread_messages (
524 message_id, thread_id, turn_id, role, text, attachments_json,
525 is_streaming, created_at, updated_at
526 ) VALUES (
527 'old-unreadable-message', 'thread-1', NULL, 'assistant',
528 'Old assistant output', 'invalid json', 0, ${now}, ${now}
529 )
530 `;
531 }),
532 );
533 }
515 if (input?.titleRegenerationBeforeStart === "two") {534 if (input?.titleRegenerationBeforeStart === "two") {
516 await Effect.runPromise(535 await Effect.runPromise(
517 engine.dispatch({536 engine.dispatch({
...@@ -561,6 +580,7 @@ describe("ProviderCommandReactor", () => {...@@ -561,6 +580,7 @@ describe("ProviderCommandReactor", () => {
561580
562 return {581 return {
563 engine,582 engine,
583 snapshotQuery,
564 readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),584 readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),
565 readPendingTurnStarts: () =>585 readPendingTurnStarts: () =>
566 runtime!.runPromise(586 runtime!.runPromise(
...@@ -3533,8 +3553,8 @@ describe("ProviderCommandReactor", () => {...@@ -3533,8 +3553,8 @@ describe("ProviderCommandReactor", () => {
3533 });3553 });
3534 });3554 });
35353555
3536 it("reacts to thread.approval.respond by forwarding provider approval response", async () => {3556 it("forwards approval responses without reading unrelated message bodies", async () => {
3537 const harness = await createHarness();3557 const harness = await createHarness({ unreadableHistory: true });
3538 const now = "2026-01-01T00:00:00.000Z";3558 const now = "2026-01-01T00:00:00.000Z";
35393559
3540 await Effect.runPromise(3560 await Effect.runPromise(
...@@ -3566,7 +3586,7 @@ describe("ProviderCommandReactor", () => {...@@ -3566,7 +3586,7 @@ describe("ProviderCommandReactor", () => {
3566 }),3586 }),
3567 );3587 );
35683588
3569 await waitFor(() => harness.respondToRequest.mock.calls.length === 1);3589 await harness.drain();
3570 expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({3590 expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({
3571 threadId: "thread-1",3591 threadId: "thread-1",
3572 requestId: "approval-request-1",3592 requestId: "approval-request-1",
...@@ -3574,8 +3594,8 @@ describe("ProviderCommandReactor", () => {...@@ -3574,8 +3594,8 @@ describe("ProviderCommandReactor", () => {
3574 });3594 });
3575 });3595 });
35763596
3577 it("reacts to thread.user-input.respond by forwarding structured user input answers", async () => {3597 it("forwards user input answers without reading unrelated message bodies", async () => {
3578 const harness = await createHarness();3598 const harness = await createHarness({ unreadableHistory: true });
3579 const now = "2026-01-01T00:00:00.000Z";3599 const now = "2026-01-01T00:00:00.000Z";
35803600
3581 await Effect.runPromise(3601 await Effect.runPromise(
...@@ -3609,7 +3629,7 @@ describe("ProviderCommandReactor", () => {...@@ -3609,7 +3629,7 @@ describe("ProviderCommandReactor", () => {
3609 }),3629 }),
3610 );3630 );
36113631
3612 await waitFor(() => harness.respondToUserInput.mock.calls.length === 1);3632 await harness.drain();
3613 expect(harness.respondToUserInput.mock.calls[0]?.[0]).toEqual({3633 expect(harness.respondToUserInput.mock.calls[0]?.[0]).toEqual({
3614 threadId: "thread-1",3634 threadId: "thread-1",
3615 requestId: "user-input-request-1",3635 requestId: "user-input-request-1",
...@@ -3823,12 +3843,12 @@ describe("ProviderCommandReactor", () => {...@@ -3823,12 +3843,12 @@ describe("ProviderCommandReactor", () => {
3823 expect(resolvedActivity).toBeUndefined();3843 expect(resolvedActivity).toBeUndefined();
3824 });3844 });
38253845
3826 it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => {3846 effectIt.effect("stops a provider session without reading unrelated message bodies", () =>
3827 const harness = await createHarness();3847 Effect.gen(function* () {
3828 const now = "2026-01-01T00:00:00.000Z";3848 const harness = yield* Effect.promise(() => createHarness({ unreadableHistory: true }));
3849 const now = "2026-01-01T00:00:00.000Z";
38293850
3830 await harness.runEffect(3851 yield* harness.engine.dispatch({
3831 harness.engine.dispatch({
3832 type: "thread.session.set",3852 type: "thread.session.set",
3833 commandId: CommandId.make("cmd-session-set-for-stop"),3853 commandId: CommandId.make("cmd-session-set-for-stop"),
3834 threadId: ThreadId.make("thread-1"),3854 threadId: ThreadId.make("thread-1"),
...@@ -3843,27 +3863,27 @@ describe("ProviderCommandReactor", () => {...@@ -3843,27 +3863,27 @@ describe("ProviderCommandReactor", () => {
3843 updatedAt: now,3863 updatedAt: now,
3844 },3864 },
3845 createdAt: now,3865 createdAt: now,
3846 }),3866 });
3847 );
38483867
3849 await harness.runEffect(3868 yield* harness.engine.dispatch({
3850 harness.engine.dispatch({
3851 type: "thread.session.stop",3869 type: "thread.session.stop",
3852 commandId: CommandId.make("cmd-session-stop"),3870 commandId: CommandId.make("cmd-session-stop"),
3853 threadId: ThreadId.make("thread-1"),3871 threadId: ThreadId.make("thread-1"),
3854 createdAt: now,3872 createdAt: now,
3855 }),3873 });
3856 );
38573874
3858 await waitFor(() => harness.stopSession.mock.calls.length === 1);3875 yield* Effect.promise(() => harness.drain());
3859 const readModel = await harness.readModel();3876 expect(harness.stopSession).toHaveBeenCalledWith({ threadId: ThreadId.make("thread-1") });
3860 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));3877 const thread = yield* harness.snapshotQuery
3861 expect(thread?.session).not.toBeNull();3878 .getThreadShellById(ThreadId.make("thread-1"))
3862 expect(thread?.session?.status).toBe("stopped");3879 .pipe(Effect.map(Option.getOrThrow));
3863 expect(thread?.session?.threadId).toBe("thread-1");3880 expect(thread.session).not.toBeNull();
3864 expect(thread?.session?.providerInstanceId).toBe(ProviderInstanceId.make("codex_work"));3881 expect(thread.session?.status).toBe("stopped");
3865 expect(thread?.session?.activeTurnId).toBeNull();3882 expect(thread.session?.threadId).toBe("thread-1");
3866 });3883 expect(thread.session?.providerInstanceId).toBe(ProviderInstanceId.make("codex_work"));
3884 expect(thread.session?.activeTurnId).toBeNull();
3885 }),
3886 );
38673887
3868 effectIt.effect("stops a ready provider session after automatic settlement", () =>3888 effectIt.effect("stops a ready provider session after automatic settlement", () =>
3869 Effect.gen(function* () {3889 Effect.gen(function* () {
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+23-17
...@@ -419,7 +419,7 @@ const make = Effect.gen(function* () {...@@ -419,7 +419,7 @@ const make = Effect.gen(function* () {
419 readonly detail: string;419 readonly detail: string;
420 readonly createdAt: string;420 readonly createdAt: string;
421 }) {421 }) {
422 const thread = yield* resolveThread(input.threadId);422 const thread = yield* resolveThreadShell(input.threadId);
423 if (!thread) {423 if (!thread) {
424 return;424 return;
425 }425 }
...@@ -447,7 +447,7 @@ const make = Effect.gen(function* () {...@@ -447,7 +447,7 @@ const make = Effect.gen(function* () {
447 compactingThreadIds.delete(threadId);447 compactingThreadIds.delete(threadId);
448 return;448 return;
449 }449 }
450 const thread = yield* resolveThread(threadId);450 const thread = yield* resolveThreadShell(threadId);
451 if (!thread?.session) return;451 if (!thread?.session) return;
452 if (452 if (
453 thread.session.status !== "starting" &&453 thread.session.status !== "starting" &&
...@@ -525,7 +525,13 @@ const make = Effect.gen(function* () {...@@ -525,7 +525,13 @@ const make = Effect.gen(function* () {
525 );525 );
526 });526 });
527527
528 const resolveThread = Effect.fnUntraced(function* (threadId: ThreadId) {528 const resolveThreadShell = Effect.fnUntraced(function* (threadId: ThreadId) {
529 return yield* projectionSnapshotQuery
530 .getThreadShellById(threadId)
531 .pipe(Effect.map(Option.getOrUndefined));
532 });
533
534 const resolveThreadDetail = Effect.fnUntraced(function* (threadId: ThreadId) {
529 return yield* projectionSnapshotQuery535 return yield* projectionSnapshotQuery
530 .getThreadDetailById(threadId, { activityKinds: [] })536 .getThreadDetailById(threadId, { activityKinds: [] })
531 .pipe(Effect.map(Option.getOrUndefined));537 .pipe(Effect.map(Option.getOrUndefined));
...@@ -571,7 +577,7 @@ const make = Effect.gen(function* () {...@@ -571,7 +577,7 @@ const make = Effect.gen(function* () {
571 readonly pendingTurnStart?: boolean;577 readonly pendingTurnStart?: boolean;
572 },578 },
573 ) {579 ) {
574 const thread = yield* resolveThread(threadId);580 const thread = yield* resolveThreadShell(threadId);
575 if (!thread) {581 if (!thread) {
576 return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`));582 return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`));
577 }583 }
...@@ -834,7 +840,7 @@ const make = Effect.gen(function* () {...@@ -834,7 +840,7 @@ const make = Effect.gen(function* () {
834 readonly interactionMode?: "default" | "plan";840 readonly interactionMode?: "default" | "plan";
835 readonly createdAt: string;841 readonly createdAt: string;
836 }) {842 }) {
837 const thread = yield* resolveThread(input.threadId);843 const thread = yield* resolveThreadShell(input.threadId);
838 if (!thread) {844 if (!thread) {
839 return yield* Effect.die(845 return yield* Effect.die(
840 new Error(`Thread '${input.threadId}' was not found in read model.`),846 new Error(`Thread '${input.threadId}' was not found in read model.`),
...@@ -975,7 +981,7 @@ const make = Effect.gen(function* () {...@@ -975,7 +981,7 @@ const make = Effect.gen(function* () {
975 );981 );
976 if (!generated) return;982 if (!generated) return;
977983
978 const thread = yield* resolveThread(input.threadId);984 const thread = yield* resolveThreadShell(input.threadId);
979 if (!thread) return;985 if (!thread) return;
980 if (!canReplaceThreadTitle(thread.title, input.titleSeed)) {986 if (!canReplaceThreadTitle(thread.title, input.titleSeed)) {
981 return;987 return;
...@@ -1007,7 +1013,7 @@ const make = Effect.gen(function* () {...@@ -1007,7 +1013,7 @@ const make = Effect.gen(function* () {
1007 return { _tag: "Superseded" } as const;1013 return { _tag: "Superseded" } as const;
1008 }1014 }
10091015
1010 const thread = yield* resolveThread(event.payload.threadId);1016 const thread = yield* resolveThreadDetail(event.payload.threadId);
1011 if (!thread || thread.titleRegeneration?.requestId !== requestId) {1017 if (!thread || thread.titleRegeneration?.requestId !== requestId) {
1012 return { _tag: "Superseded" } as const;1018 return { _tag: "Superseded" } as const;
1013 }1019 }
...@@ -1040,7 +1046,7 @@ const make = Effect.gen(function* () {...@@ -1040,7 +1046,7 @@ const make = Effect.gen(function* () {
1040 return { _tag: "Completed", title: undefined } as const;1046 return { _tag: "Completed", title: undefined } as const;
1041 }1047 }
10421048
1043 const latestThread = yield* resolveThread(event.payload.threadId);1049 const latestThread = yield* resolveThreadShell(event.payload.threadId);
1044 if (1050 if (
1045 !latestThread ||1051 !latestThread ||
1046 latestThread.titleRegeneration?.requestId !== requestId ||1052 latestThread.titleRegeneration?.requestId !== requestId ||
...@@ -1177,7 +1183,7 @@ const make = Effect.gen(function* () {...@@ -1177,7 +1183,7 @@ const make = Effect.gen(function* () {
1177 return;1183 return;
1178 }1184 }
11791185
1180 const thread = yield* resolveThread(event.payload.threadId);1186 const thread = yield* resolveThreadDetail(event.payload.threadId);
1181 if (!thread) {1187 if (!thread) {
1182 return;1188 return;
1183 }1189 }
...@@ -1366,7 +1372,7 @@ const make = Effect.gen(function* () {...@@ -1366,7 +1372,7 @@ const make = Effect.gen(function* () {
1366 "Context compaction requires an existing conversation.",1372 "Context compaction requires an existing conversation.",
1367 );1373 );
1368 }1374 }
1369 const latestThread = yield* resolveThread(event.payload.threadId);1375 const latestThread = yield* resolveThreadShell(event.payload.threadId);
1370 if (1376 if (
1371 compactingThreadIds.has(event.payload.threadId) ||1377 compactingThreadIds.has(event.payload.threadId) ||
1372 latestThread?.session?.status === "starting" ||1378 latestThread?.session?.status === "starting" ||
...@@ -1436,7 +1442,7 @@ const make = Effect.gen(function* () {...@@ -1436,7 +1442,7 @@ const make = Effect.gen(function* () {
1436 const processTurnInterruptRequested = Effect.fn("processTurnInterruptRequested")(function* (1442 const processTurnInterruptRequested = Effect.fn("processTurnInterruptRequested")(function* (
1437 event: Extract<ProviderIntentEvent, { type: "thread.turn-interrupt-requested" }>,1443 event: Extract<ProviderIntentEvent, { type: "thread.turn-interrupt-requested" }>,
1438 ) {1444 ) {
1439 const thread = yield* resolveThread(event.payload.threadId);1445 const thread = yield* resolveThreadShell(event.payload.threadId);
1440 if (!thread) {1446 if (!thread) {
1441 return;1447 return;
1442 }1448 }
...@@ -1459,7 +1465,7 @@ const make = Effect.gen(function* () {...@@ -1459,7 +1465,7 @@ const make = Effect.gen(function* () {
14591465
1460 const detail = formatFailureDetail(cause);1466 const detail = formatFailureDetail(cause);
1461 return Effect.gen(function* () {1467 return Effect.gen(function* () {
1462 const latestThread = yield* resolveThread(event.payload.threadId);1468 const latestThread = yield* resolveThreadShell(event.payload.threadId);
1463 const latestSession = latestThread?.session;1469 const latestSession = latestThread?.session;
1464 if (1470 if (
1465 !latestSession ||1471 !latestSession ||
...@@ -1487,7 +1493,7 @@ const make = Effect.gen(function* () {...@@ -1487,7 +1493,7 @@ const make = Effect.gen(function* () {
1487 );1493 );
1488 }),1494 }),
1489 );1495 );
1490 const stoppedThread = yield* resolveThread(event.payload.threadId);1496 const stoppedThread = yield* resolveThreadShell(event.payload.threadId);
1491 const stoppedSession = stoppedThread?.session;1497 const stoppedSession = stoppedThread?.session;
1492 if (1498 if (
1493 !stoppedSession ||1499 !stoppedSession ||
...@@ -1531,7 +1537,7 @@ const make = Effect.gen(function* () {...@@ -1531,7 +1537,7 @@ const make = Effect.gen(function* () {
1531 const processApprovalResponseRequested = Effect.fn("processApprovalResponseRequested")(function* (1537 const processApprovalResponseRequested = Effect.fn("processApprovalResponseRequested")(function* (
1532 event: Extract<ProviderIntentEvent, { type: "thread.approval-response-requested" }>,1538 event: Extract<ProviderIntentEvent, { type: "thread.approval-response-requested" }>,
1533 ) {1539 ) {
1534 const thread = yield* resolveThread(event.payload.threadId);1540 const thread = yield* resolveThreadShell(event.payload.threadId);
1535 if (!thread) {1541 if (!thread) {
1536 return;1542 return;
1537 }1543 }
...@@ -1575,7 +1581,7 @@ const make = Effect.gen(function* () {...@@ -1575,7 +1581,7 @@ const make = Effect.gen(function* () {
1575 function* (1581 function* (
1576 event: Extract<ProviderIntentEvent, { type: "thread.user-input-response-requested" }>,1582 event: Extract<ProviderIntentEvent, { type: "thread.user-input-response-requested" }>,
1577 ) {1583 ) {
1578 const thread = yield* resolveThread(event.payload.threadId);1584 const thread = yield* resolveThreadShell(event.payload.threadId);
1579 if (!thread) {1585 if (!thread) {
1580 return;1586 return;
1581 }1587 }
...@@ -1619,7 +1625,7 @@ const make = Effect.gen(function* () {...@@ -1619,7 +1625,7 @@ const make = Effect.gen(function* () {
1619 const processSessionStopRequested = Effect.fn("processSessionStopRequested")(function* (1625 const processSessionStopRequested = Effect.fn("processSessionStopRequested")(function* (
1620 event: Extract<ProviderIntentEvent, { type: "thread.session-stop-requested" }>,1626 event: Extract<ProviderIntentEvent, { type: "thread.session-stop-requested" }>,
1621 ) {1627 ) {
1622 const thread = yield* resolveThread(event.payload.threadId);1628 const thread = yield* resolveThreadShell(event.payload.threadId);
1623 if (!thread) {1629 if (!thread) {
1624 return;1630 return;
1625 }1631 }
...@@ -1696,7 +1702,7 @@ const make = Effect.gen(function* () {...@@ -1696,7 +1702,7 @@ const make = Effect.gen(function* () {
1696 yield* threadTitleRegenerationWorker.enqueue(event);1702 yield* threadTitleRegenerationWorker.enqueue(event);
1697 return;1703 return;
1698 case "thread.runtime-mode-set": {1704 case "thread.runtime-mode-set": {
1699 const thread = yield* resolveThread(event.payload.threadId);1705 const thread = yield* resolveThreadShell(event.payload.threadId);
1700 if (!thread?.session || thread.session.status === "stopped") {1706 if (!thread?.session || thread.session.status === "stopped") {
1701 return;1707 return;
1702 }1708 }