diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index dcd08d8a2586995a95b1ddeb1c2f16848f187ed6..8449e06cb6537a2b735c87da7822ec013a2d5b08 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -890,6 +890,203 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("caps a SQL window by runs that a notification started", () => + Effect.gen(function* () { + // A thread woken by PR watches: 12 user turns, then 400 runs that a + // notification started, with a user turn every 50 runs. The user-turn + // window reached back to the 12th-newest user turn and read almost every row. + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const nowIso = DateTime.formatIso(now); + const threadId = ThreadId.make("thread:notification-turn-pages"); + yield* projectionStore.apply({ + id: EventId.make("event:notification-turn-pages:thread"), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:notification-turn-pages"), + title: "Woken by notifications", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + + const startedRuns = 412; + const allIds: string[] = []; + const startIds: string[] = []; + let ordinal = 0; + const itemRow = (runId: string, item: Record) => { + ordinal += 1; + const id = `turn-item:notification-turn-pages:${ordinal}`; + allIds.push(id); + return { + turn_item_id: id, + thread_id: threadId, + run_id: runId, + node_id: null, + provider_thread_id: null, + provider_turn_id: null, + parent_item_id: null, + ordinal, + type: String(item.type), + status: "completed", + updated_at: nowIso, + payload_json: encodeUnknownJsonString({ + id, + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: nowIso, + completedAt: nowIso, + updatedAt: nowIso, + ...item, + }), + }; + }; + const command = { type: "command_execution", input: "command", output: "x", exitCode: 0 }; + const notification = (summary: string) => ({ + type: "notification", + source: { kind: "monitor" }, + outcome: "completed", + summary, + }); + // The newest run is a wake queued behind the last one. It has no items yet, + // like a run cancelled before it started. + for (let run = 1; run <= startedRuns + 1; run += 1) { + const runId = `run:notification-turn-pages:${run}`; + const status = run > startedRuns ? "queued" : "completed"; + yield* sql` + INSERT INTO orchestration_v2_projection_runs ( + run_id, thread_id, ordinal, provider, provider_thread_id, status, + requested_at, completed_at, payload_json + ) VALUES ( + ${runId}, ${threadId}, ${run}, 'codex', NULL, ${status}, ${nowIso}, ${nowIso}, + ${encodeUnknownJsonString({ + id: runId, + threadId, + ordinal: run, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: `message:notification-turn-pages:${run}`, + rootNodeId: `node:notification-turn-pages:${run}`, + activeAttemptId: null, + status, + requestedAt: nowIso, + startedAt: nowIso, + completedAt: nowIso, + checkpointId: null, + contextHandoffId: null, + })} + ) + `; + if (run > startedRuns) continue; + const start = + run <= 12 || run % 50 === 0 + ? { + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId: `message:notification-turn-pages:${run}`, + inputIntent: "turn_start", + text: `Turn ${run}`, + attachments: [], + } + : notification(`Wake ${run}`); + // A row of the run (a handoff) can come before its first input. A wake + // steered into the running turn follows it and starts no turn. + const rows = [ + itemRow(runId, command), + itemRow(runId, start), + itemRow(runId, command), + itemRow(runId, notification(`Steer ${run}`)), + ]; + startIds.push(rows[1]!.turn_item_id); + yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`; + } + + const { statements, tracer } = traceSqlStatements(); + const initial = yield* projectionStore + .getThreadSnapshotWindow(threadId, { rowLimit: 77, userTurnLimit: 10 }) + .pipe(Effect.withTracer(tracer)); + // The 152 newest turns are the 152 newest started runs, 261 to 412. Only + // three of them are user turns, short of the 12 the user-turn window needs. + assert.deepEqual( + initial.projection.turnItems.map((item) => String(item.id)), + allIds.slice(allIds.indexOf(startIds[260]!)), + ); + // Each run's first input is read by run ID, newest run first, never by a + // walk over every turn item in the thread. + const windowStatement = statements.find((statement) => statement.includes("run_starts")); + assert.isDefined(windowStatement); + const windowPlan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${windowStatement}`, + ); + const windowPlanDetails = windowPlan.map((row) => row.detail); + assert.include( + windowPlanDetails, + "SEARCH recent_run USING INDEX orchestration_v2_projection_runs_thread_ordinal_idx (thread_id=? AND ordinal String(row.sourceItemId)); + let cursor = bounded.historyCursor; + for (let pages = 0; cursor !== null; pages += 1) { + assert.isBelow(pages, 5); + const anchor = decodeThreadHistoryCursor(cursor); + const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 77, + userTurnLimit: 20, + anchorItemId: TurnItemId.make(anchor.si), + anchorThreadId: ThreadId.make(anchor.st), + }); + // Older windows stay bounded too. + assert.isAtMost(snapshot.projection.turnItems.length, 152 * 4 + 1); + const page = selectHistoryPageFromCursor({ + items: snapshot.projection.visibleTurnItems, + cursor, + snapshotSequence: 0, + }); + loaded.unshift(...page.items.map((row) => String(row.sourceItemId))); + cursor = page.nextCursor; + } + assert.deepEqual(loaded, allIds); + }), + ); + it.effect( "bounds completed nodes within one long run while preserving ancestry and live work", () => diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 11f228f4041803affdddd09e6299d913b84eb26c..44c7c79579174305b66c310287907c36732876f7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -85,7 +85,7 @@ import { threadHtmlRenderAttachmentIds } from "../attachmentStore.ts"; import { isThreadHistoryUserTurn, isConversationHistoryItem, - isThreadHistoryTurnStart, + threadHistoryTurnStarts, THREAD_HISTORY_MAX_RAW_TURNS, selectHistoryPageFromCursor, selectRecentTimelineWindow, @@ -2824,12 +2824,54 @@ export const layer: Layer.Layer = AND request.type = 'run_interrupt_request' ) ) + ), run_starts AS ( + -- A wake (PR watch, task completion) starts its run with a + -- notification instead of a user message. Read the first + -- input of each run, newest run first, and add the + -- notifications to the user-message turn starts. A run + -- without an eligible first input (queued, rolled back) is + -- skipped, so it cannot hide older turns. CROSS JOIN keeps + -- the runs as the outer loop, and each probe reads only the + -- first rows of a run. + SELECT start.ordinal, start.type + FROM orchestration_v2_projection_runs AS recent_run + CROSS JOIN eligible AS start ON start.item_rowid = ( + SELECT first_input.rowid + FROM orchestration_v2_projection_turn_items AS first_input + WHERE first_input.run_id = recent_run.run_id + AND first_input.type IN ('user_message', 'notification') + ORDER BY first_input.ordinal + LIMIT 1 + ) + WHERE recent_run.thread_id = ${threadId} + AND recent_run.ordinal <= COALESCE( + ( + SELECT anchor_run.ordinal + FROM orchestration_v2_projection_turn_items AS anchor_item + JOIN orchestration_v2_projection_runs AS anchor_run + ON anchor_run.run_id = anchor_item.run_id + WHERE anchor_item.thread_id = ${threadId} + AND anchor_item.turn_item_id = ${window.anchorItemId ?? null} + LIMIT 1 + ), + 9223372036854775807 + ) + AND ${window.userTurnLimit ?? null} IS NOT NULL + ORDER BY recent_run.ordinal DESC + LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2} ), turn_anchors AS ( SELECT ordinal, payload_json - FROM eligible - WHERE type = 'user_message' - AND json_extract(payload_json, '$.inputIntent') IN ('turn_start', 'queued_turn') - AND ${window.userTurnLimit ?? null} IS NOT NULL + FROM ( + SELECT ordinal, payload_json + FROM eligible + WHERE type = 'user_message' + AND json_extract(payload_json, '$.inputIntent') IN ('turn_start', 'queued_turn') + AND ${window.userTurnLimit ?? null} IS NOT NULL + ORDER BY ordinal DESC + LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2} + ) + UNION ALL + SELECT ordinal, NULL FROM run_starts WHERE type = 'notification' ORDER BY ordinal DESC LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2} ), user_anchors AS ( @@ -6679,12 +6721,11 @@ export const layerMemory: Layer.Layer = Layer.effect( (row) => row.sourceItemId === options.anchorItemId, ) + 1; const candidates = snapshot.projection.visibleTurnItems.slice(0, anchorIndex); + const turnStarts = threadHistoryTurnStarts(candidates.map((row) => row.item)); const turnAnchors = options.userTurnLimit === undefined ? [] - : candidates.flatMap((row, index) => - isThreadHistoryTurnStart(row.item) ? [index] : [], - ); + : candidates.flatMap((_row, index) => (turnStarts[index] ? [index] : [])); const rawStart = turnAnchors.at(-(THREAD_HISTORY_MAX_RAW_TURNS + 2)) ?? 0; const anchors = turnAnchors.filter( (index) => index >= rawStart && isThreadHistoryUserTurn(candidates[index]!.item), diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 57a6bf86da587167185fac51d53a1ec3c1c9c302..da694288c6054c03f1c0e5978f193158f768b8f7 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -209,6 +209,57 @@ describe("threadHistoryPaging", () => { ); }); + it("counts runs that a notification started toward the 150-turn ceiling", () => { + // A wake (PR watch, task completion) starts its run with a notification. A + // row of the run (a handoff) can come before it, and a wake steered into the + // running turn comes after it and starts no turn. + const inRun = (index: number, run: number, item: object = {}) => { + const row = makeRow(index); + return { + ...row, + item: { ...row.item, runId: RunId.make(`run-${run}`), ...item }, + } as OrchestrationV2ProjectedTurnItem; + }; + const notification = (summary: string) => ({ + type: "notification", + source: { kind: "monitor" }, + outcome: "completed", + summary, + }); + const items = Array.from({ length: 161 }, (_, run) => [ + inRun(run * 4, run), + inRun( + run * 4 + 1, + run, + run === 0 + ? { + type: "user_message", + createdBy: "user", + creationSource: "web", + inputIntent: "turn_start", + messageId: MessageId.make("prompt-0"), + text: "Prompt 0", + attachments: [], + } + : notification(`Wake ${run}`), + ), + inRun(run * 4 + 2, run, notification(`Steer ${run}`)), + inRun(run * 4 + 3, run), + ]).flat(); + const first = selectRecentTimelineWindow({ items, snapshotSequence: 1 }); + // 150 runs from the start of run 11. Its earlier row goes to the older page. + expect(first.items).toHaveLength(150 * 4 - 1); + expect(first.items[0]?.sourceItemId).toBe(`item-${11 * 4 + 1}`); + const older = selectHistoryPageFromCursor({ + items, + cursor: first.nextCursor!, + snapshotSequence: 1, + }); + expect([...older.items, ...first.items].map((row) => row.sourceItemId)).toEqual( + items.map((row) => row.sourceItemId), + ); + }); + it("pages agent-only child transcripts instead of dropping their earlier activity", () => { const commandRows = Array.from({ length: 90 }, (_, index) => makeRow(index + 1)); const first = makeRow(0); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index ce7fccfd7d26ce0c53791208c80bf7b3002c71e3..c4a1b38e256c6d390fa150a7e5feaaba24315505 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -53,6 +53,7 @@ type HistoryRow = Pick< readonly messageId?: string; readonly inputIntent?: string; readonly createdBy?: string; + readonly runId?: string | null; }; }; export type SelectTimelinePageResult = { @@ -162,13 +163,32 @@ export function decodeThreadHistoryCursor(cursor: string): ThreadHistoryCursorPa return parsed as ThreadHistoryCursorPayload; } -export function isThreadHistoryTurnStart(item: HistoryRow["item"]): boolean { +function isThreadHistoryTurnStart(item: HistoryRow["item"]): boolean { return ( item.type === "user_message" && (item.inputIntent === "turn_start" || item.inputIntent === "queued_turn") ); } +/** + * Marks the turn starts of a chronological timeline. A wake (PR watch, task + * completion) starts its run with a notification instead of a user message, so + * a notification starts a turn when it is its run's first input. One steered + * into a running turn comes later. ProjectionStore's `run_starts` window query + * applies the same rule. + */ +export function threadHistoryTurnStarts( + items: ReadonlyArray, +): ReadonlyArray { + const startedRuns = new Set(); + return items.map((item) => { + if (item.type !== "user_message" && item.type !== "notification") return false; + const firstInput = item.runId != null && !startedRuns.has(item.runId); + if (item.runId != null) startedRuns.add(item.runId); + return item.type === "notification" ? firstInput : isThreadHistoryTurnStart(item); + }); +} + /** Steering belongs to its existing turn and must not consume another page slot. */ export function isThreadHistoryUserTurn(item: HistoryRow["item"]): boolean { return ( @@ -199,7 +219,9 @@ function selectOlderTimelinePage(input: { let encodedBytes = 0; let userTurns = 0; let rawTurns = 0; - const turnLimit = input.items.slice(0, end).some((row) => isThreadHistoryUserTurn(row.item)) + const candidates = input.items.slice(0, end); + const turnStarts = threadHistoryTurnStarts(candidates.map((row) => row.item)); + const turnLimit = candidates.some((row) => isThreadHistoryUserTurn(row.item)) ? policy.maxUserTurns : undefined; for (let index = end - 1; index >= 0; index -= 1) { @@ -216,7 +238,7 @@ function selectOlderTimelinePage(input: { selected.push(row); encodedBytes += rowBytes; if (isThreadHistoryUserTurn(row.item)) userTurns += 1; - if (isThreadHistoryTurnStart(row.item)) rawTurns += 1; + if (turnStarts[index]) rawTurns += 1; } selected.reverse();