From 1c5637adbdba11f46a3876f13b0e212b0f7fa98a Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Tue, 27 Oct 2026 13:37:34 +0000 Subject: [PATCH 1/6] fix(server): pending questions stay answerable after their turn leaves the history window A Codex question or approval left pending while later turns completed kept its runtime request (so the sidebar showed Input), but the thread snapshot window dropped the turn holding its card. Clients read the prompt from that card, so the conversation had nothing to answer or dismiss. The SQL snapshot window and the bounded snapshot now retain cards of pending requests in turnItems, outside the visible rows, like interrupt requests. Paging older history shows a retained card in place. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 202 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 44 ++-- .../src/state/threadHistoryMerge.test.ts | 46 ++++ .../src/state/threadHistoryMerge.ts | 6 +- 4 files changed, 281 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 9bd3566e559a..47bc11902c1b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1555,6 +1555,208 @@ it.effect("SQL list older settled threads below a cursor", () => }).pipe(Effect.provide(layerTest)), ); +// regression: #25806 pending request cards outlive the history window that drops their turn +it.effect.each([ + ["memory", ProjectionStore.layerMemory], + ["SQL", layerTest], +] as const)("%s snapshot windows keep cards for requests still pending", ([, layer]) => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:pending-request-window"); + yield* store.apply({ + id: EventId.make("event:pending-request-window:thread"), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:pending-request-window"), + title: "Pending request window", + 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, + }, + }); + let ordinal = 0; + const itemBase = (runId: RunId, nodeId: NodeId, id: string) => ({ + id: TurnItemId.make(`item:pending-request-window:${id}`), + threadId, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: ++ordinal, + status: "completed" as const, + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + }); + const request = (id: string, nodeId: NodeId, kind: "user_input" | "command") => ({ + id: RuntimeRequestId.make(`request:pending-request-window:${id}`), + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind, + responseCapability: { + type: "live" as const, + providerSessionId: ProviderSessionId.make("session:pending-request-window"), + }, + createdAt: now, + }); + const turnCount = THREAD_HISTORY_PAGE_POLICY.maxUserTurns + 4; + for (let turn = 1; turn <= turnCount; turn += 1) { + const runId = RunId.make(`run:pending-request-window:${turn}`); + const nodeId = NodeId.make(`node:pending-request-window:${turn}`); + yield* store.apply({ + id: EventId.make(`event:pending-request-window:run:${turn}`), + type: "run.created", + threadId, + runId, + nodeId, + driver, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal: turn, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:pending-request-window:${turn}`), + rootNodeId: nodeId, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + const items: Array = [ + { + ...itemBase(runId, nodeId, `message:${turn}`), + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId: MessageId.make(`message:pending-request-window:${turn}`), + inputIntent: "turn_start", + text: `Turn ${turn}`, + attachments: [], + }, + ]; + if (turn === 1) { + const question = { + id: "scope", + header: "Scope", + question: "Which scope?", + options: [{ label: "All", description: "Everything" }], + }; + items.push( + { + ...itemBase(runId, nodeId, "question:pending"), + status: "running", + type: "user_input_request", + requestId: request("question:pending", nodeId, "user_input").id, + questions: [question], + }, + { + ...itemBase(runId, nodeId, "approval:pending"), + status: "running", + type: "approval_request", + requestId: request("approval:pending", nodeId, "command").id, + requestKind: "command", + prompt: "rm -rf build", + }, + { + ...itemBase(runId, nodeId, "question:resolved"), + type: "user_input_request", + requestId: request("question:resolved", nodeId, "user_input").id, + questions: [question], + }, + ); + for (const [id, kind, status] of [ + ["question:pending", "user_input", "pending"], + ["approval:pending", "command", "pending"], + ["question:resolved", "user_input", "resolved"], + ] as const) { + yield* store.apply({ + id: EventId.make(`event:pending-request-window:request:${id}`), + type: "runtime-request.updated", + threadId, + nodeId, + driver, + occurredAt: now, + payload: { + ...request(id, nodeId, kind), + status, + resolvedAt: status === "pending" ? null : now, + }, + }); + } + } + for (const item of items) { + yield* store.apply({ + id: EventId.make(`event:pending-request-window:${item.id}`), + type: "turn-item.updated", + threadId, + runId, + nodeId, + driver, + occurredAt: now, + payload: item, + }); + } + } + + const pendingCardIds = [ + TurnItemId.make("item:pending-request-window:question:pending"), + TurnItemId.make("item:pending-request-window:approval:pending"), + ]; + const snapshot = yield* store.getThreadSnapshotWindow(threadId, { + rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }); + const bounded = buildBoundedThreadProjection({ + projection: snapshot.projection, + snapshotSequence: snapshot.snapshotSequence, + }); + assert.isTrue(bounded.hasMoreHistory); + assert.notInclude( + bounded.projection.turnItems.map((item) => item.id), + TurnItemId.make("item:pending-request-window:question:resolved"), + ); + for (const projection of [snapshot.projection, bounded.projection]) { + const visibleIds = projection.visibleTurnItems.map((row) => row.sourceItemId); + assert.notInclude(visibleIds, TurnItemId.make("item:pending-request-window:message:1")); + assert.includeMembers( + projection.turnItems.map((item) => item.id), + pendingCardIds, + ); + for (const id of pendingCardIds) assert.notInclude(visibleIds, id); + } + }).pipe(Effect.provide(layer)), +); + it.effect("lists only active subagent IDs without loading shell data", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 95c0f7a4bae5..b9f1a91d7c9a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -2967,21 +2967,6 @@ export const layer: Layer.Layer = ) AND request.type = 'run_interrupt_request' UNION - -- An open question or approval stays in the window: clients - -- build the answer panel from its row, and a run that kept - -- working pushes the row out of the newest turns. - SELECT question.rowid AS item_rowid, question.ordinal, question.turn_item_id - FROM orchestration_v2_projection_runtime_requests AS pending - JOIN orchestration_v2_projection_turn_items AS question - ON question.node_id = pending.node_id - WHERE pending.thread_id = ${threadId} - AND pending.status = 'pending' - AND question.thread_id = ${threadId} - AND question.type IN ('user_input_request', 'approval_request') - AND ${window.rowLimit} > 0 - AND ${window.anchorItemId ?? null} IS NULL - AND ${window.requiredRunId ?? null} IS NULL - UNION SELECT latest.item_rowid, latest.ordinal, latest.turn_item_id FROM ( SELECT rowid AS item_rowid, ordinal, turn_item_id @@ -5822,6 +5807,33 @@ export const layer: Layer.Layer = : {}), ...(historyAnchor === undefined ? {} : { historyAnchor }), }); + // Pending requests outlive their turn's window; clients read their prompt from the card. + const retainedIds = new Set(projection.turnItems.map((item) => item.id)); + const pendingCards = (yield* decodeRows( + decodeTurnItemPayload, + threadId, + )( + yield* sql` + SELECT item.payload_json + FROM orchestration_v2_projection_runtime_requests AS request + JOIN orchestration_v2_projection_turn_items AS item + ON item.node_id = request.node_id + AND item.type IN ('approval_request', 'user_input_request') + AND json_extract(item.payload_json, '$.requestId') = request.runtime_request_id + WHERE request.thread_id = ${threadId} AND request.status = 'pending' + AND item.thread_id = ${threadId} + `, + )).filter((item) => !retainedIds.has(item.id)); + const windowProjection = + pendingCards.length === 0 + ? projection + : { + ...projection, + turnItems: [...projection.turnItems, ...pendingCards].toSorted( + (left, right) => + left.ordinal - right.ordinal || left.id.localeCompare(right.id), + ), + }; const rows = yield* sql<{ readonly snapshot_sequence: number | null }>` SELECT MAX(sequence) AS snapshot_sequence FROM orchestration_events @@ -5832,7 +5844,7 @@ export const layer: Layer.Layer = return { schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION, snapshotSequence: rows[0]?.snapshot_sequence ?? 0, - projection, + projection: windowProjection, }; }), ) diff --git a/packages/client-runtime/src/state/threadHistoryMerge.test.ts b/packages/client-runtime/src/state/threadHistoryMerge.test.ts index 59e1ea3652e5..98bbc0c980fb 100644 --- a/packages/client-runtime/src/state/threadHistoryMerge.test.ts +++ b/packages/client-runtime/src/state/threadHistoryMerge.test.ts @@ -222,6 +222,52 @@ describe("threadHistoryMerge", () => { expect(next?.visibleTurnItems[3]?.item).toEqual(liveItem); }); + // regression: #25806 a pending question card retained outside the window shows when paged in + it("shows a pending request card retained at cold open when its history page arrives", () => { + const recent = row(5); + const card = { + id: "item-question" as never, + type: "user_input_request" as const, + threadId: v2ThreadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running" as const, + title: null, + requestId: "request-question" as never, + questions: [], + startedAt: NOW, + completedAt: null, + updatedAt: NOW, + }; + const coldOpen = { + ...v2Projection, + turnItems: [card, recent.item], + visibleTurnItems: [{ ...recent, position: 0 }], + }; + + const merged = mergeOlderHistoryIntoProjection(coldOpen, [ + { ...row(0), position: 0 }, + { + position: 1, + visibility: "local", + sourceThreadId: v2ThreadId, + sourceItemId: card.id, + item: card, + }, + ]); + + expect(merged.visibleTurnItems.map((entry) => String(entry.sourceItemId))).toEqual([ + "item-0", + "item-question", + "item-5", + ]); + }); + it("keeps a history-page interrupt result visible when its request was retained at cold open", () => { const runId = "run-interrupt-cold" as never; const nodeId = "node-interrupt-cold" as never; diff --git a/packages/client-runtime/src/state/threadHistoryMerge.ts b/packages/client-runtime/src/state/threadHistoryMerge.ts index 6fd678b66b12..dcec2f207b81 100644 --- a/packages/client-runtime/src/state/threadHistoryMerge.ts +++ b/packages/client-runtime/src/state/threadHistoryMerge.ts @@ -89,9 +89,13 @@ export function mergeOlderHistoryIntoProjection( (row.visibility === "local" || row.sourceThreadId === projection.thread.id) ) { const currentItem = turnItemById.get(String(row.sourceItemId)); + // Hidden live items stay hidden. Snapshots also retain interrupt requests and + // pending request cards outside the visible window; their history page shows them. if (currentItem !== undefined) { if ( - currentItem.type !== "run_interrupt_request" || + (currentItem.type !== "run_interrupt_request" && + currentItem.type !== "approval_request" && + currentItem.type !== "user_input_request") || !isOrchestrationV2TurnItemVisible({ item: currentItem, runs: projection.runs, From b994cf38603aec32e5ef9d1a00698b5e08ca5b0f Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Tue, 27 Oct 2026 14:06:53 +0000 Subject: [PATCH 2/6] fix: recover cached questions and index pending card lookups --- .../orchestration-v2/ProjectionStore.test.ts | 33 +++++-- .../src/orchestration-v2/ProjectionStore.ts | 2 +- .../src/state/threads-sync.test.ts | 87 ++++++++++++++++++- packages/client-runtime/src/state/threads.ts | 29 ++++++- 4 files changed, 141 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 47bc11902c1b..a15063765e62 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1559,7 +1559,7 @@ it.effect("SQL list older settled threads below a cursor", () => it.effect.each([ ["memory", ProjectionStore.layerMemory], ["SQL", layerTest], -] as const)("%s snapshot windows keep cards for requests still pending", ([, layer]) => +] as const)("%s snapshot windows keep cards for requests still pending", ([storeKind, layer]) => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; @@ -1732,10 +1732,33 @@ it.effect.each([ TurnItemId.make("item:pending-request-window:question:pending"), TurnItemId.make("item:pending-request-window:approval:pending"), ]; - const snapshot = yield* store.getThreadSnapshotWindow(threadId, { - rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2, - userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, - }); + const { statements, tracer } = traceSqlStatements(); + const snapshot = yield* store + .getThreadSnapshotWindow(threadId, { + rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }) + .pipe(Effect.withTracer(tracer)); + if (storeKind === "SQL") { + const sql = yield* SqlClient.SqlClient; + const query = statements.find( + (statement) => + statement.includes("FROM orchestration_v2_projection_runtime_requests AS request") && + statement.includes("SELECT item.payload_json"), + ); + assert.isDefined(query); + const plan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${query}`, + )).map((row) => row.detail); + assert.include( + plan[0], + "SEARCH request USING INDEX orchestration_v2_projection_runtime_requests_thread_status_idx", + ); + assert.include( + plan[1], + "SEARCH item USING INDEX orchestration_v2_projection_turn_items_node_ordinal_idx", + ); + } const bounded = buildBoundedThreadProjection({ projection: snapshot.projection, snapshotSequence: snapshot.snapshotSequence, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b9f1a91d7c9a..7754212ab978 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5816,7 +5816,7 @@ export const layer: Layer.Layer = yield* sql` SELECT item.payload_json FROM orchestration_v2_projection_runtime_requests AS request - JOIN orchestration_v2_projection_turn_items AS item + CROSS JOIN orchestration_v2_projection_turn_items AS item ON item.node_id = request.node_id AND item.type IN ('approval_request', 'user_input_request') AND json_extract(item.payload_json, '$.requestId') = request.runtime_request_id diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 593e29286b66..47e16cb8c839 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -317,6 +317,32 @@ const PENDING_QUESTION = { resolvedAt: null, }; +const QUESTION_CARD: OrchestrationV2TurnItem = { + id: TurnItemId.make("question-card"), + threadId: THREAD_ID, + runId: null, + nodeId: PENDING_QUESTION.nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: null, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + type: "user_input_request", + requestId: PENDING_QUESTION.id, + questions: [{ id: "scope", header: "Scope", question: "Which scope?", options: [] }], +}; + +const PROJECTION_WITH_QUESTION = { + ...BASE_PROJECTION, + runtimeRequests: [PENDING_QUESTION], + turnItems: [QUESTION_CARD], +}; + const questionResolved = (sequence: number): OrchestrationV2ThreadStreamItem => ({ kind: "event", sequence, @@ -360,6 +386,63 @@ const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => { }; describe("EnvironmentThreads", () => { + it.effect.each([ + ["disk", false], + ["retained", false], + ["disk", true], + ["retained", true], + ] as const)( + "recovers a %s cache only when its pending question card is missing: %s", + ([cacheKind, hasCard]) => + Effect.gen(function* () { + const projection = { + ...PROJECTION_WITH_QUESTION, + turnItems: hasCard ? [QUESTION_CARD] : [], + }; + const schema = Schema.fromJsonString(StoredOrchestrationThreadSnapshot); + const encoded = yield* Schema.encodeEffect(schema)({ + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + environmentId: TARGET.environmentId, + threadId: THREAD_ID, + snapshot: { snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, projection }, + }); + const stored = yield* Schema.decodeEffect(schema)(encoded); + const h = yield* makeHarness({ + loadCached: Effect.succeedSome(stored.snapshot), + ...(cacheKind === "retained" + ? { + resumeCache: { + owner: undefined, + snapshot: { + state: { + data: Option.some(projection), + status: "live", + error: Option.none(), + history: EMPTY_THREAD_HISTORY_META, + }, + sequence: CACHED_SNAPSHOT_SEQUENCE, + persisted: true, + }, + }, + } + : {}), + httpSnapshot: { + _tag: "present", + snapshot: { + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + projection: PROJECTION_WITH_QUESTION, + }, + }, + completionMarker: true, + }); + yield* Queue.offer(h.inputs, synchronized()); + const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(yield* Ref.get(h.loaderCalls)).toBe(hasCard ? 0 : 1); + expect(Option.getOrThrow(state.data)).toEqual(PROJECTION_WITH_QUESTION); + expect(yield* Ref.get(h.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE); + }), + ); + it.effect("discards an oversized version-4 cache and requests a fresh bounded snapshot", () => Effect.gen(function* () { const item: OrchestrationV2TurnItem = { @@ -559,7 +642,7 @@ describe("EnvironmentThreads", () => { Scope.close(scope, Exit.void), ); const old = yield* makeHarness({ - cached: { ...BASE_PROJECTION, runtimeRequests: [PENDING_QUESTION] }, + cached: PROJECTION_WITH_QUESTION, resumeCache, }).pipe(Effect.provideService(Scope.Scope, oldScope)); yield* awaitThreadState(old.observed, (value) => value.status === "live"); @@ -2146,7 +2229,7 @@ describe("EnvironmentThreads", () => { (unsubscribe) => Effect.sync(unsubscribe), ); const harness = yield* makeHarness({ - cached: { ...BASE_PROJECTION, runtimeRequests: [PENDING_QUESTION] }, + cached: PROJECTION_WITH_QUESTION, completionMarker: true, }); yield* awaitThreadState( diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 5c6622b5ad9a..b13570c9f3c1 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -188,6 +188,20 @@ function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadStat }; } +function hasMissingPendingQuestionCard(projection: OrchestrationV2ThreadProjection): boolean { + const pending = new Set( + projection.runtimeRequests + .filter((request) => request.kind === "user_input" && request.status === "pending") + .map((request) => request.id), + ); + if (pending.size === 0) return false; + for (const item of projection.turnItems) { + if (item.type === "user_input_request") pending.delete(item.requestId); + if (pending.size === 0) return false; + } + return true; +} + export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, resumeCache?: ThreadResumeCache, @@ -205,10 +219,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; - const retained = resumeCache?.snapshot; + const candidate = resumeCache?.snapshot; + const retained = + candidate && + Option.isSome(candidate.state.data) && + hasMissingPendingQuestionCard(candidate.state.data.value) + ? undefined + : candidate; const owner = {}; if (resumeCache) resumeCache.owner = owner; - const cached = + const loaded = retained === undefined ? yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => @@ -223,6 +243,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ), ) : Option.none(); + // Old bounded caches can omit a pending question's card; replay alone cannot restore it. + const cached = Option.filter( + loaded, + (snapshot) => !hasMissingPendingQuestionCard(snapshot.projection), + ); const cachedThread = Option.map(cached, (snapshot) => snapshot.projection); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body. From 027bd20883ab0e302e97639afd81fa5e440bd878 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Wed, 28 Oct 2026 10:48:26 +0000 Subject: [PATCH 3/6] test(server): make snapshot query-plan assertions optional for memory stores --- apps/server/src/orchestration-v2/ProjectionStore.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a15063765e62..91862567449d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -31,6 +31,7 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; @@ -1739,8 +1740,9 @@ it.effect.each([ userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, }) .pipe(Effect.withTracer(tracer)); + const sqlClient = yield* Effect.serviceOption(SqlClient.SqlClient); if (storeKind === "SQL") { - const sql = yield* SqlClient.SqlClient; + const sql = Option.getOrThrow(sqlClient); const query = statements.find( (statement) => statement.includes("FROM orchestration_v2_projection_runtime_requests AS request") && From 80d2e4b76799550948df97d6c722e9a4fa36b0c9 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Wed, 28 Oct 2026 11:34:22 +0000 Subject: [PATCH 4/6] test(server): use SQLite for pending card regression --- .../orchestration-v2/ProjectionStore.test.ts | 47 ++++++++----------- 1 file changed, 20 insertions(+), 27 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 91862567449d..6286935984ba 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -31,7 +31,6 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; -import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; @@ -1557,10 +1556,7 @@ it.effect("SQL list older settled threads below a cursor", () => ); // regression: #25806 pending request cards outlive the history window that drops their turn -it.effect.each([ - ["memory", ProjectionStore.layerMemory], - ["SQL", layerTest], -] as const)("%s snapshot windows keep cards for requests still pending", ([storeKind, layer]) => +it.effect("SQL snapshot windows keep cards for requests still pending", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; @@ -1740,27 +1736,24 @@ it.effect.each([ userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, }) .pipe(Effect.withTracer(tracer)); - const sqlClient = yield* Effect.serviceOption(SqlClient.SqlClient); - if (storeKind === "SQL") { - const sql = Option.getOrThrow(sqlClient); - const query = statements.find( - (statement) => - statement.includes("FROM orchestration_v2_projection_runtime_requests AS request") && - statement.includes("SELECT item.payload_json"), - ); - assert.isDefined(query); - const plan = (yield* sql.unsafe<{ readonly detail: string }>( - `EXPLAIN QUERY PLAN ${query}`, - )).map((row) => row.detail); - assert.include( - plan[0], - "SEARCH request USING INDEX orchestration_v2_projection_runtime_requests_thread_status_idx", - ); - assert.include( - plan[1], - "SEARCH item USING INDEX orchestration_v2_projection_turn_items_node_ordinal_idx", - ); - } + const sql = yield* SqlClient.SqlClient; + const query = statements.find( + (statement) => + statement.includes("FROM orchestration_v2_projection_runtime_requests AS request") && + statement.includes("SELECT item.payload_json"), + ); + assert.isDefined(query); + const plan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${query}`, + )).map((row) => row.detail); + assert.include( + plan[0], + "SEARCH request USING INDEX orchestration_v2_projection_runtime_requests_thread_status_idx", + ); + assert.include( + plan[1], + "SEARCH item USING INDEX orchestration_v2_projection_turn_items_node_ordinal_idx", + ); const bounded = buildBoundedThreadProjection({ projection: snapshot.projection, snapshotSequence: snapshot.snapshotSequence, @@ -1779,7 +1772,7 @@ it.effect.each([ ); for (const id of pendingCardIds) assert.notInclude(visibleIds, id); } - }).pipe(Effect.provide(layer)), + }).pipe(Effect.provide(layerTest)), ); it.effect("lists only active subagent IDs without loading shell data", () => From 8316f5a29da63a59a421c5357cea84ff32032775 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Thu, 29 Oct 2026 16:40:48 +0000 Subject: [PATCH 5/6] test(client-runtime): build retained cache fixture state from the empty thread state Spreading EMPTY_ENVIRONMENT_THREAD_STATE keeps the fixture valid when EnvironmentThreadState gains fields such as sequence. Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/state/threads-sync.test.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 47e16cb8c839..15560867adfd 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -415,10 +415,9 @@ describe("EnvironmentThreads", () => { owner: undefined, snapshot: { state: { + ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(projection), status: "live", - error: Option.none(), - history: EMPTY_THREAD_HISTORY_META, }, sequence: CACHED_SNAPSHOT_SEQUENCE, persisted: true, From bcc2c460ec1e60a6802260976513aa3704cb3f9c Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Thu, 29 Oct 2026 17:09:05 +0000 Subject: [PATCH 6/6] fix(server): reserve snapshot space for pending question cards --- .../threadHistoryPaging.test.ts | 64 +++++++++ .../orchestration-v2/threadHistoryPaging.ts | 5 +- .../src/state/threadHistoryMerge.test.ts | 130 ++++++++++++------ 3 files changed, 154 insertions(+), 45 deletions(-) diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index f4e54000222f..3430828d2871 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -2,6 +2,7 @@ import { MessageId, NodeId, RunId, + RuntimeRequestId, ThreadId, TurnItemId, type OrchestrationV2ProjectedTurnItem, @@ -738,6 +739,69 @@ describe("threadHistoryPaging", () => { ).toBe(2_097_152); }); + it.each([40_000, 80_000])( + "reserves pending question cards before visible history: %s bytes", + (questionBytes) => { + const rows = Array.from({ length: 20 }, (_, index) => + makeRow(index + 1, { outputBytes: 1_500 }), + ); + const requestId = RuntimeRequestId.make("pending-question"); + const card: OrchestrationV2TurnItem = { + id: TurnItemId.make("pending-question-card"), + type: "user_input_request", + threadId: THREAD, + runId: null, + nodeId: NODE, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + requestId, + responseMode: "message", + questions: [ + { id: "scope", header: "Scope", question: "q".repeat(questionBytes), options: [] }, + ], + startedAt: NOW, + completedAt: NOW, + updatedAt: NOW, + }; + const projection = { + ...makeProjection(rows), + turnItems: [card, ...rows.map((row) => row.item)], + runtimeRequests: [ + { + id: requestId, + nodeId: NODE, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input" as const, + status: "pending" as const, + responseCapability: { type: "message" as const }, + createdAt: NOW, + resolvedAt: null, + }, + ], + }; + const policy = { ...THREAD_HISTORY_PAGE_POLICY, maxEncodedBytes: 64_000 }; + + const bounded = buildBoundedThreadProjection({ projection, snapshotSequence: 9, policy }); + + expect(bounded.projection.turnItems).toContain(card); + expect(bounded.projection.visibleTurnItems.length).toBeGreaterThan(0); + expect(bounded.hasMoreHistory).toBe(true); + if (questionBytes < policy.maxEncodedBytes) { + expect(Buffer.byteLength(JSON.stringify(bounded.projection), "utf8")).toBeLessThanOrEqual( + policy.maxEncodedBytes, + ); + } else { + expect(bounded.projection.visibleTurnItems).toHaveLength(1); + } + }, + ); + it("preserves oversized actionable state", () => { const projection = makeProjection([makeRow(0)]); const actionable = { diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 03d056531a76..35f262c652db 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -673,10 +673,9 @@ export function buildBoundedThreadProjection(input: { turnItems: [], visibleTurnItems: [], }); - // Control state (checkpoint file lists, the latest task list) cannot be paged, so it may claim at most - // half the budget; past that the window would shrink to the one row every page admits. + // Keep the control-only window minimum, but required dependency cards claim their reserve first. const windowBudget = Math.max( - policy.maxEncodedBytes / 2, + dependencyReserve === 0 ? policy.maxEncodedBytes / 2 : 0, policy.maxEncodedBytes - dependencyReserve - controlBytes - 1_024, ); const windowPolicy: ThreadHistoryPagePolicy = { ...policy, maxEncodedBytes: windowBudget }; diff --git a/packages/client-runtime/src/state/threadHistoryMerge.test.ts b/packages/client-runtime/src/state/threadHistoryMerge.test.ts index 98bbc0c980fb..3f2be832a2b2 100644 --- a/packages/client-runtime/src/state/threadHistoryMerge.test.ts +++ b/packages/client-runtime/src/state/threadHistoryMerge.test.ts @@ -5,6 +5,7 @@ import * as DateTime from "effect/DateTime"; import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; import { mergeOlderHistoryIntoProjection } from "./threadHistoryMerge.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; +import { derivePendingThreadRequests } from "./threadRequests.ts"; const NOW = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"); @@ -223,50 +224,95 @@ describe("threadHistoryMerge", () => { }); // regression: #25806 a pending question card retained outside the window shows when paged in - it("shows a pending request card retained at cold open when its history page arrives", () => { - const recent = row(5); - const card = { - id: "item-question" as never, - type: "user_input_request" as const, - threadId: v2ThreadId, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "running" as const, - title: null, - requestId: "request-question" as never, - questions: [], - startedAt: NOW, - completedAt: null, - updatedAt: NOW, - }; - const coldOpen = { - ...v2Projection, - turnItems: [card, recent.item], - visibleTurnItems: [{ ...recent, position: 0 }], - }; + it.each(["completed", "rolled_back"] as const)( + "keeps a pending question actionable when paging a %s run", + (status) => { + const recent = row(5); + const runId = "question-run" as never; + const nodeId = "question-node" as never; + const card = { + id: "item-question" as never, + type: "user_input_request" as const, + threadId: v2ThreadId, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running" as const, + title: null, + requestId: "request-question" as never, + responseMode: "message" as const, + questions: [{ id: "scope", header: "Scope", question: "Which scope?", options: [] }], + startedAt: NOW, + completedAt: null, + updatedAt: NOW, + }; + const coldOpen = { + ...v2Projection, + runs: [ + { + id: runId, + threadId: v2ThreadId, + ordinal: 1, + providerInstanceId: v2Projection.thread.providerInstanceId, + modelSelection: v2Projection.thread.modelSelection, + providerThreadId: null, + userMessageId: "question-message" as never, + rootNodeId: nodeId, + activeAttemptId: null, + status, + requestedAt: NOW, + startedAt: NOW, + completedAt: NOW, + checkpointId: null, + contextHandoffId: null, + }, + ], + runtimeRequests: [ + { + id: card.requestId, + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input" as const, + status: "pending" as const, + responseCapability: { type: "message" as const }, + createdAt: NOW, + resolvedAt: null, + }, + ], + turnItems: [card, recent.item], + visibleTurnItems: [{ ...recent, position: 0 }], + }; - const merged = mergeOlderHistoryIntoProjection(coldOpen, [ - { ...row(0), position: 0 }, - { - position: 1, - visibility: "local", - sourceThreadId: v2ThreadId, - sourceItemId: card.id, - item: card, - }, - ]); + const merged = mergeOlderHistoryIntoProjection(coldOpen, [ + { ...row(0), position: 0 }, + { + position: 1, + visibility: "local", + sourceThreadId: v2ThreadId, + sourceItemId: card.id, + item: card, + }, + ]); - expect(merged.visibleTurnItems.map((entry) => String(entry.sourceItemId))).toEqual([ - "item-0", - "item-question", - "item-5", - ]); - }); + expect(merged.visibleTurnItems.map((entry) => String(entry.sourceItemId))).toEqual( + status === "rolled_back" ? ["item-0", "item-5"] : ["item-0", "item-question", "item-5"], + ); + expect(merged.turnItems).toContain(card); + const pending = derivePendingThreadRequests(merged).userInputs; + expect(pending).toEqual(derivePendingThreadRequests(coldOpen).userInputs); + expect(pending).toHaveLength(1); + expect(pending[0]).toMatchObject({ + requestId: card.requestId, + dismissible: true, + questions: card.questions.map((question) => ({ ...question, multiSelect: false })), + }); + }, + ); it("keeps a history-page interrupt result visible when its request was retained at cold open", () => { const runId = "run-interrupt-cold" as never;