From d530b529562e1ae0a2196bea53a259c2bb31fd94 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 6 Oct 2026 03:10:02 -0500 Subject: [PATCH] fix(server): avoid scanning completed history for pending secrets --- .../orchestration-v2/ProjectionStore.test.ts | 67 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 10 +-- 2 files changed, 73 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 692161ad1a5d..f47b4d6829e8 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2850,6 +2850,73 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("selects the latest waiting secret only from active runs", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const suffix = "shell-pending-secret"; + const threadId = yield* addRolledBackRecoveryCandidate(suffix); + const runId = RunId.make(`run:${suffix}:rolled-back`); + const nodeId = NodeId.make(`node:${suffix}:rolled-back`); + const now = yield* DateTime.now; + const addSecret = (id: string, ordinal: number, status: "waiting" | "completed") => + projectionStore.apply({ + id: EventId.make(`event:${id}`), + type: "turn-item.updated", + threadId, + runId, + nodeId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make(id), + threadId, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status, + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "secret_request", + label: "Test credential", + reason: "Test pending input", + secretStatus: status === "waiting" ? "pending" : "saved", + }, + }); + yield* addSecret("secret:a", 2, "waiting"); + yield* addSecret("secret:b", 3, "waiting"); + yield* addSecret("secret:c", 4, "completed"); + + for (const status of [ + "preparing", + "starting", + "running", + "waiting", + "completed", + "rolled_back", + ]) { + yield* sql` + UPDATE orchestration_v2_projection_runs + SET status = ${status}, payload_json = json_set(payload_json, '$.status', ${status}) + WHERE run_id = ${runId} + `; + const shell = yield* projectionStore.getShellSnapshot(); + const thread = shell.threads.find((candidate) => candidate.id === threadId); + assert.isDefined(thread); + assert.equal( + thread?.pendingRuntimeRequest?.id ?? null, + status === "completed" || status === "rolled_back" ? null : "secret:b", + ); + } + }), + ); + it.effect("builds shell snapshots without decoding full turn item payloads", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index a700f557cfbf..ec75937a9a62 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4999,12 +4999,14 @@ export const layer: Layer.Layer = ) AS pending_request_payload_json, ( SELECT secret.payload_json - FROM orchestration_v2_projection_turn_items secret + -- Keep runs outermost so completed history is never scanned for a pending secret. + FROM orchestration_v2_projection_runs r + CROSS JOIN orchestration_v2_projection_turn_items secret INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx - INNER JOIN orchestration_v2_projection_runs r ON r.run_id = secret.run_id - WHERE secret.thread_id = t.thread_id - AND secret.type = 'secret_request' AND secret.status = 'waiting' + WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') + AND secret.thread_id = t.thread_id AND secret.run_id = r.run_id + AND secret.type = 'secret_request' AND secret.status = 'waiting' ORDER BY secret.updated_at DESC, secret.turn_item_id DESC LIMIT 1 ) AS pending_secret_request_payload_json,