authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 05:55:58-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 05:55:58-07:00
logd0a73b56c21ca1b7f39ab28d0fe46db99eec8f67
tree8b77d451b4ddcbd247a50e364d920e26b83e0b28
parent28732b4787261609c4d20b82b244cb6eef0c442e
signature Signed by PGP key B5690EEEBB952194

fix(server): threads woken by notifications no longer load their whole history (#19423)

Co-authored-by: Eldar Politkin <32969822+EldarPolitkin@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

4 files changed, 322 insertions(+), 11 deletions(-)

apps/server/src/orchestration-v2/ProjectionStore.test.ts+197
......@@ -890,6 +890,203 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
890890 }),
891891 );
892892
893 it.effect("caps a SQL window by runs that a notification started", () =>
894 Effect.gen(function* () {
895 // A thread woken by PR watches: 12 user turns, then 400 runs that a
896 // notification started, with a user turn every 50 runs. The user-turn
897 // window reached back to the 12th-newest user turn and read almost every row.
898 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
899 const sql = yield* SqlClient.SqlClient;
900 const now = yield* DateTime.now;
901 const nowIso = DateTime.formatIso(now);
902 const threadId = ThreadId.make("thread:notification-turn-pages");
903 yield* projectionStore.apply({
904 id: EventId.make("event:notification-turn-pages:thread"),
905 type: "thread.created",
906 threadId,
907 occurredAt: now,
908 payload: {
909 createdBy: "user",
910 creationSource: "web",
911 id: threadId,
912 projectId: ProjectId.make("project:notification-turn-pages"),
913 title: "Woken by notifications",
914 providerInstanceId,
915 modelSelection,
916 runtimeMode: "full-access",
917 interactionMode: "default",
918 branch: null,
919 worktreePath: null,
920 activeProviderThreadId: null,
921 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
922 forkedFrom: null,
923 createdAt: now,
924 updatedAt: now,
925 archivedAt: null,
926 settledOverride: null,
927 settledAt: null,
928 lastVisitedAt: null,
929 deletedAt: null,
930 },
931 });
932
933 const startedRuns = 412;
934 const allIds: string[] = [];
935 const startIds: string[] = [];
936 let ordinal = 0;
937 const itemRow = (runId: string, item: Record<string, unknown>) => {
938 ordinal += 1;
939 const id = `turn-item:notification-turn-pages:${ordinal}`;
940 allIds.push(id);
941 return {
942 turn_item_id: id,
943 thread_id: threadId,
944 run_id: runId,
945 node_id: null,
946 provider_thread_id: null,
947 provider_turn_id: null,
948 parent_item_id: null,
949 ordinal,
950 type: String(item.type),
951 status: "completed",
952 updated_at: nowIso,
953 payload_json: encodeUnknownJsonString({
954 id,
955 threadId,
956 runId,
957 nodeId: null,
958 providerThreadId: null,
959 providerTurnId: null,
960 nativeItemRef: null,
961 parentItemId: null,
962 ordinal,
963 status: "completed",
964 title: null,
965 startedAt: nowIso,
966 completedAt: nowIso,
967 updatedAt: nowIso,
968 ...item,
969 }),
970 };
971 };
972 const command = { type: "command_execution", input: "command", output: "x", exitCode: 0 };
973 const notification = (summary: string) => ({
974 type: "notification",
975 source: { kind: "monitor" },
976 outcome: "completed",
977 summary,
978 });
979 // The newest run is a wake queued behind the last one. It has no items yet,
980 // like a run cancelled before it started.
981 for (let run = 1; run <= startedRuns + 1; run += 1) {
982 const runId = `run:notification-turn-pages:${run}`;
983 const status = run > startedRuns ? "queued" : "completed";
984 yield* sql`
985 INSERT INTO orchestration_v2_projection_runs (
986 run_id, thread_id, ordinal, provider, provider_thread_id, status,
987 requested_at, completed_at, payload_json
988 ) VALUES (
989 ${runId}, ${threadId}, ${run}, 'codex', NULL, ${status}, ${nowIso}, ${nowIso},
990 ${encodeUnknownJsonString({
991 id: runId,
992 threadId,
993 ordinal: run,
994 providerInstanceId,
995 modelSelection,
996 providerThreadId: null,
997 userMessageId: `message:notification-turn-pages:${run}`,
998 rootNodeId: `node:notification-turn-pages:${run}`,
999 activeAttemptId: null,
1000 status,
1001 requestedAt: nowIso,
1002 startedAt: nowIso,
1003 completedAt: nowIso,
1004 checkpointId: null,
1005 contextHandoffId: null,
1006 })}
1007 )
1008 `;
1009 if (run > startedRuns) continue;
1010 const start =
1011 run <= 12 || run % 50 === 0
1012 ? {
1013 type: "user_message",
1014 createdBy: "user",
1015 creationSource: "web",
1016 messageId: `message:notification-turn-pages:${run}`,
1017 inputIntent: "turn_start",
1018 text: `Turn ${run}`,
1019 attachments: [],
1020 }
1021 : notification(`Wake ${run}`);
1022 // A row of the run (a handoff) can come before its first input. A wake
1023 // steered into the running turn follows it and starts no turn.
1024 const rows = [
1025 itemRow(runId, command),
1026 itemRow(runId, start),
1027 itemRow(runId, command),
1028 itemRow(runId, notification(`Steer ${run}`)),
1029 ];
1030 startIds.push(rows[1]!.turn_item_id);
1031 yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`;
1032 }
1033
1034 const { statements, tracer } = traceSqlStatements();
1035 const initial = yield* projectionStore
1036 .getThreadSnapshotWindow(threadId, { rowLimit: 77, userTurnLimit: 10 })
1037 .pipe(Effect.withTracer(tracer));
1038 // The 152 newest turns are the 152 newest started runs, 261 to 412. Only
1039 // three of them are user turns, short of the 12 the user-turn window needs.
1040 assert.deepEqual(
1041 initial.projection.turnItems.map((item) => String(item.id)),
1042 allIds.slice(allIds.indexOf(startIds[260]!)),
1043 );
1044 // Each run's first input is read by run ID, newest run first, never by a
1045 // walk over every turn item in the thread.
1046 const windowStatement = statements.find((statement) => statement.includes("run_starts"));
1047 assert.isDefined(windowStatement);
1048 const windowPlan = yield* sql.unsafe<{ readonly detail: string }>(
1049 `EXPLAIN QUERY PLAN ${windowStatement}`,
1050 );
1051 const windowPlanDetails = windowPlan.map((row) => row.detail);
1052 assert.include(
1053 windowPlanDetails,
1054 "SEARCH recent_run USING INDEX orchestration_v2_projection_runs_thread_ordinal_idx (thread_id=? AND ordinal<?)",
1055 );
1056 assert.include(
1057 windowPlanDetails,
1058 "SEARCH first_input USING INDEX orchestration_v2_projection_turn_items_run_ordinal_idx (run_id=?)",
1059 );
1060 const bounded = buildBoundedThreadProjection({
1061 projection: initial.projection,
1062 snapshotSequence: 0,
1063 });
1064 assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, startIds[262]);
1065 const loaded = bounded.projection.visibleTurnItems.map((row) => String(row.sourceItemId));
1066 let cursor = bounded.historyCursor;
1067 for (let pages = 0; cursor !== null; pages += 1) {
1068 assert.isBelow(pages, 5);
1069 const anchor = decodeThreadHistoryCursor(cursor);
1070 const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1071 rowLimit: 77,
1072 userTurnLimit: 20,
1073 anchorItemId: TurnItemId.make(anchor.si),
1074 anchorThreadId: ThreadId.make(anchor.st),
1075 });
1076 // Older windows stay bounded too.
1077 assert.isAtMost(snapshot.projection.turnItems.length, 152 * 4 + 1);
1078 const page = selectHistoryPageFromCursor({
1079 items: snapshot.projection.visibleTurnItems,
1080 cursor,
1081 snapshotSequence: 0,
1082 });
1083 loaded.unshift(...page.items.map((row) => String(row.sourceItemId)));
1084 cursor = page.nextCursor;
1085 }
1086 assert.deepEqual(loaded, allIds);
1087 }),
1088 );
1089
8931090 it.effect(
8941091 "bounds completed nodes within one long run while preserving ancestry and live work",
8951092 () =>
apps/server/src/orchestration-v2/ProjectionStore.ts+49-8
......@@ -85,7 +85,7 @@ import { threadHtmlRenderAttachmentIds } from "../attachmentStore.ts";
8585import {
8686 isThreadHistoryUserTurn,
8787 isConversationHistoryItem,
88 isThreadHistoryTurnStart,
88 threadHistoryTurnStarts,
8989 THREAD_HISTORY_MAX_RAW_TURNS,
9090 selectHistoryPageFromCursor,
9191 selectRecentTimelineWindow,
......@@ -2824,12 +2824,54 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
28242824 AND request.type = 'run_interrupt_request'
28252825 )
28262826 )
2827 ), run_starts AS (
2828 -- A wake (PR watch, task completion) starts its run with a
2829 -- notification instead of a user message. Read the first
2830 -- input of each run, newest run first, and add the
2831 -- notifications to the user-message turn starts. A run
2832 -- without an eligible first input (queued, rolled back) is
2833 -- skipped, so it cannot hide older turns. CROSS JOIN keeps
2834 -- the runs as the outer loop, and each probe reads only the
2835 -- first rows of a run.
2836 SELECT start.ordinal, start.type
2837 FROM orchestration_v2_projection_runs AS recent_run
2838 CROSS JOIN eligible AS start ON start.item_rowid = (
2839 SELECT first_input.rowid
2840 FROM orchestration_v2_projection_turn_items AS first_input
2841 WHERE first_input.run_id = recent_run.run_id
2842 AND first_input.type IN ('user_message', 'notification')
2843 ORDER BY first_input.ordinal
2844 LIMIT 1
2845 )
2846 WHERE recent_run.thread_id = ${threadId}
2847 AND recent_run.ordinal <= COALESCE(
2848 (
2849 SELECT anchor_run.ordinal
2850 FROM orchestration_v2_projection_turn_items AS anchor_item
2851 JOIN orchestration_v2_projection_runs AS anchor_run
2852 ON anchor_run.run_id = anchor_item.run_id
2853 WHERE anchor_item.thread_id = ${threadId}
2854 AND anchor_item.turn_item_id = ${window.anchorItemId ?? null}
2855 LIMIT 1
2856 ),
2857 9223372036854775807
2858 )
2859 AND ${window.userTurnLimit ?? null} IS NOT NULL
2860 ORDER BY recent_run.ordinal DESC
2861 LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2}
28272862 ), turn_anchors AS (
28282863 SELECT ordinal, payload_json
2829 FROM eligible
2830 WHERE type = 'user_message'
2831 AND json_extract(payload_json, '$.inputIntent') IN ('turn_start', 'queued_turn')
2832 AND ${window.userTurnLimit ?? null} IS NOT NULL
2864 FROM (
2865 SELECT ordinal, payload_json
2866 FROM eligible
2867 WHERE type = 'user_message'
2868 AND json_extract(payload_json, '$.inputIntent') IN ('turn_start', 'queued_turn')
2869 AND ${window.userTurnLimit ?? null} IS NOT NULL
2870 ORDER BY ordinal DESC
2871 LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2}
2872 )
2873 UNION ALL
2874 SELECT ordinal, NULL FROM run_starts WHERE type = 'notification'
28332875 ORDER BY ordinal DESC
28342876 LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2}
28352877 ), user_anchors AS (
......@@ -6679,12 +6721,11 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
66796721 (row) => row.sourceItemId === options.anchorItemId,
66806722 ) + 1;
66816723 const candidates = snapshot.projection.visibleTurnItems.slice(0, anchorIndex);
6724 const turnStarts = threadHistoryTurnStarts(candidates.map((row) => row.item));
66826725 const turnAnchors =
66836726 options.userTurnLimit === undefined
66846727 ? []
6685 : candidates.flatMap((row, index) =>
6686 isThreadHistoryTurnStart(row.item) ? [index] : [],
6687 );
6728 : candidates.flatMap((_row, index) => (turnStarts[index] ? [index] : []));
66886729 const rawStart = turnAnchors.at(-(THREAD_HISTORY_MAX_RAW_TURNS + 2)) ?? 0;
66896730 const anchors = turnAnchors.filter(
66906731 (index) => index >= rawStart && isThreadHistoryUserTurn(candidates[index]!.item),
apps/server/src/orchestration-v2/threadHistoryPaging.test.ts+51
......@@ -209,6 +209,57 @@ describe("threadHistoryPaging", () => {
209209 );
210210 });
211211
212 it("counts runs that a notification started toward the 150-turn ceiling", () => {
213 // A wake (PR watch, task completion) starts its run with a notification. A
214 // row of the run (a handoff) can come before it, and a wake steered into the
215 // running turn comes after it and starts no turn.
216 const inRun = (index: number, run: number, item: object = {}) => {
217 const row = makeRow(index);
218 return {
219 ...row,
220 item: { ...row.item, runId: RunId.make(`run-${run}`), ...item },
221 } as OrchestrationV2ProjectedTurnItem;
222 };
223 const notification = (summary: string) => ({
224 type: "notification",
225 source: { kind: "monitor" },
226 outcome: "completed",
227 summary,
228 });
229 const items = Array.from({ length: 161 }, (_, run) => [
230 inRun(run * 4, run),
231 inRun(
232 run * 4 + 1,
233 run,
234 run === 0
235 ? {
236 type: "user_message",
237 createdBy: "user",
238 creationSource: "web",
239 inputIntent: "turn_start",
240 messageId: MessageId.make("prompt-0"),
241 text: "Prompt 0",
242 attachments: [],
243 }
244 : notification(`Wake ${run}`),
245 ),
246 inRun(run * 4 + 2, run, notification(`Steer ${run}`)),
247 inRun(run * 4 + 3, run),
248 ]).flat();
249 const first = selectRecentTimelineWindow({ items, snapshotSequence: 1 });
250 // 150 runs from the start of run 11. Its earlier row goes to the older page.
251 expect(first.items).toHaveLength(150 * 4 - 1);
252 expect(first.items[0]?.sourceItemId).toBe(`item-${11 * 4 + 1}`);
253 const older = selectHistoryPageFromCursor({
254 items,
255 cursor: first.nextCursor!,
256 snapshotSequence: 1,
257 });
258 expect([...older.items, ...first.items].map((row) => row.sourceItemId)).toEqual(
259 items.map((row) => row.sourceItemId),
260 );
261 });
262
212263 it("pages agent-only child transcripts instead of dropping their earlier activity", () => {
213264 const commandRows = Array.from({ length: 90 }, (_, index) => makeRow(index + 1));
214265 const first = makeRow(0);
apps/server/src/orchestration-v2/threadHistoryPaging.ts+25-3
......@@ -53,6 +53,7 @@ type HistoryRow = Pick<
5353 readonly messageId?: string;
5454 readonly inputIntent?: string;
5555 readonly createdBy?: string;
56 readonly runId?: string | null;
5657 };
5758};
5859export type SelectTimelinePageResult<Row extends HistoryRow = OrchestrationV2ProjectedTurnItem> = {
......@@ -162,13 +163,32 @@ export function decodeThreadHistoryCursor(cursor: string): ThreadHistoryCursorPa
162163 return parsed as ThreadHistoryCursorPayload;
163164}
164165
165export function isThreadHistoryTurnStart(item: HistoryRow["item"]): boolean {
166function isThreadHistoryTurnStart(item: HistoryRow["item"]): boolean {
166167 return (
167168 item.type === "user_message" &&
168169 (item.inputIntent === "turn_start" || item.inputIntent === "queued_turn")
169170 );
170171}
171172
173/**
174 * Marks the turn starts of a chronological timeline. A wake (PR watch, task
175 * completion) starts its run with a notification instead of a user message, so
176 * a notification starts a turn when it is its run's first input. One steered
177 * into a running turn comes later. ProjectionStore's `run_starts` window query
178 * applies the same rule.
179 */
180export function threadHistoryTurnStarts(
181 items: ReadonlyArray<HistoryRow["item"]>,
182): ReadonlyArray<boolean> {
183 const startedRuns = new Set<string>();
184 return items.map((item) => {
185 if (item.type !== "user_message" && item.type !== "notification") return false;
186 const firstInput = item.runId != null && !startedRuns.has(item.runId);
187 if (item.runId != null) startedRuns.add(item.runId);
188 return item.type === "notification" ? firstInput : isThreadHistoryTurnStart(item);
189 });
190}
191
172192/** Steering belongs to its existing turn and must not consume another page slot. */
173193export function isThreadHistoryUserTurn(item: HistoryRow["item"]): boolean {
174194 return (
......@@ -199,7 +219,9 @@ function selectOlderTimelinePage<Row extends HistoryRow>(input: {
199219 let encodedBytes = 0;
200220 let userTurns = 0;
201221 let rawTurns = 0;
202 const turnLimit = input.items.slice(0, end).some((row) => isThreadHistoryUserTurn(row.item))
222 const candidates = input.items.slice(0, end);
223 const turnStarts = threadHistoryTurnStarts(candidates.map((row) => row.item));
224 const turnLimit = candidates.some((row) => isThreadHistoryUserTurn(row.item))
203225 ? policy.maxUserTurns
204226 : undefined;
205227 for (let index = end - 1; index >= 0; index -= 1) {
......@@ -216,7 +238,7 @@ function selectOlderTimelinePage<Row extends HistoryRow>(input: {
216238 selected.push(row);
217239 encodedBytes += rowBytes;
218240 if (isThreadHistoryUserTurn(row.item)) userTurns += 1;
219 if (isThreadHistoryTurnStart(row.item)) rawTurns += 1;
241 if (turnStarts[index]) rawTurns += 1;
220242 }
221243 selected.reverse();
222244