diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index dd76e5eca01..3001a585e97 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -733,6 +733,33 @@ fn validate_pending_runtime_requests( Ok(()) } +fn attested_pending_runtime_requests( + pending: &BTreeMap, + provider: &crate::acpx_provider_state::AcpxProviderState, + live: &BTreeMap, + durable_turn_id: &str, +) -> Vec { + pending + .values() + .filter(|request| { + let id = request["requestId"].as_str().unwrap_or(""); + // Canonical requests bind the durable PRP turn. The sidecar + // separately attests the provider-assigned turn. + request["turnId"].as_str() == Some(durable_turn_id) + && if request["type"] == "input" { + provider + .pending_provider_input_request_id(id) + .and_then(|provider_id| live.get(provider_id)) + .is_some_and(|kind| kind == "input") + } else { + provider.pending_permission(id).is_some() + && live.get(id).is_some_and(|kind| kind == "permission") + } + }) + .cloned() + .collect() +} + pub struct AcpxCommandExecutor { state_dir: PathBuf, context: AcpxEventProjectionContext, @@ -1783,30 +1810,16 @@ impl AcpxCommandExecutor { )); } validate_pending_runtime_requests(&state.pending_runtime_requests)?; - let requests = state - .pending_runtime_requests - .values() - .filter(|request| { - let id = request["requestId"].as_str().unwrap_or(""); - request["turnId"].as_str() == state.active_turn_id.as_deref() - && if request["type"] == "input" { - session - .state() - .pending_provider_input_request_id(id) - .and_then(|provider_id| live_requests.get(provider_id)) - .is_some_and(|kind| kind == "input") - } else { - session.state().pending_permission(id).is_some() - && live_requests - .get(id) - .is_some_and(|kind| kind == "permission") - } - }) - .cloned() - .collect::>(); + let requests = attested_pending_runtime_requests( + &state.pending_runtime_requests, + session.state(), + &live_requests, + &self.context.turn_id, + ); let mut snapshot = self.snapshot()?; snapshot.result["pendingRuntimeRequests"] = json!(requests); snapshot.result["runtimeRequestsLive"] = json!(true); + snapshot.result["runtimeRequestTurnId"] = json!(self.context.turn_id); Ok(snapshot) } @@ -2443,6 +2456,84 @@ mod tests { "origin":{"adapter":"acpx-runtime-sidecar","provider":"cursor","method":"cursor/ask_question"}}) } + #[test] + fn live_request_snapshot_preserves_durable_turn_and_provider_callback_bindings() { + use crate::acpx_provider_state::AcpxProviderState; + use crate::acpx_sidecar_transport::AcpxSidecarEvent; + use crate::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType; + let mut provider = AcpxProviderState::new("run-1").unwrap(); + provider.begin_turn("provider-turn-1").unwrap(); + let mut projection = context(); + projection.turn_id = "durable-turn-1".into(); + projection.provider_turn_id = Some("provider-turn-1".into()); + let mut pending = BTreeMap::new(); + for (sequence, event_type, payload) in [ + ( + 1, + GeneratedAcpxSidecarEventType::RuntimeInputRequested, + json!({"requestId":"raw input / 1","questionSet":pending_input_request("unused")["input"]}), + ), + ( + 2, + GeneratedAcpxSidecarEventType::RuntimePermissionRequested, + json!({"requestId":"permission-1","kind":"execute","title":"Run?","choices":[{"key":"decline","label":"Decline"}]}), + ), + ] { + let event = AcpxSidecarEvent { + sequence, + event_type, + run_id: Some("run-1".into()), + turn_id: Some("provider-turn-1".into()), + payload, + }; + for event in provider.accept_event(&event).unwrap() { + for normalized in project_acpx_state_event(&projection, &event).unwrap() { + if normalized.event_type == "runtime_request.created" { + let request = normalized.payload["request"].clone(); + pending.insert(request["requestId"].as_str().unwrap().to_owned(), request); + } + } + } + } + validate_pending_runtime_requests(&pending).unwrap(); + let live = BTreeMap::from([ + ("raw input / 1".into(), "input".into()), + ("permission-1".into(), "permission".into()), + ]); + let requests = + attested_pending_runtime_requests(&pending, &provider, &live, "durable-turn-1"); + assert_eq!(requests.len(), 2); + assert!(requests + .iter() + .all(|request| request["turnId"] == "durable-turn-1")); + assert!(requests + .iter() + .any(|request| request["requestId"] != "raw input / 1" && request["type"] == "input")); + assert!( + attested_pending_runtime_requests(&pending, &provider, &live, "provider-turn-1") + .is_empty() + ); + assert!(attested_pending_runtime_requests( + &pending, + &provider, + &BTreeMap::new(), + "durable-turn-1" + ) + .is_empty()); + provider.complete_permission("permission-1").unwrap(); + let input_id = requests + .iter() + .find(|request| request["type"] == "input") + .unwrap()["requestId"] + .as_str() + .unwrap(); + provider.complete_input(input_id).unwrap(); + assert!( + attested_pending_runtime_requests(&pending, &provider, &live, "durable-turn-1") + .is_empty() + ); + } + #[test] fn durable_runtime_ledger_retains_only_unsettled_requests_and_rejects_forged_types() { let operations = Vec::new(); diff --git a/packages/paperclip-runner/src/live/acpx-request-recovery.test.ts b/packages/paperclip-runner/src/live/acpx-request-recovery.test.ts index ed84c5d0d73..2e2fd43b9c8 100644 --- a/packages/paperclip-runner/src/live/acpx-request-recovery.test.ts +++ b/packages/paperclip-runner/src/live/acpx-request-recovery.test.ts @@ -6,17 +6,17 @@ import type { CodexRpcServerRequest } from "../drivers/codex/app-server-transpor function snapshot() { return { provider: "acpx", runtimeRequestsLive: true, - driverSessionId: "thread-1", activeProviderTurnId: "turn-1", + driverSessionId: "thread-1", activeProviderTurnId: "turn-1", runtimeRequestTurnId: "durable-turn-1", pendingRuntimeRequests: [{ schema: "paperclip.runtime_request.v2", requestId: "input-1", type: "input", - requestKind: "runtime", status: "pending", turnId: "turn-1", itemId: "question-1", + requestKind: "runtime", status: "pending", turnId: "durable-turn-1", itemId: "question-1", origin: { adapter: "acpx-runtime-sidecar", provider: "pi", method: "elicitation/create" }, input: { schema: "paperclip.question_set.v1", questions: [ { id: "answer", prompt: "Give the answer", required: true, answerMode: "text" }, ] }, }, { schema: "paperclip.runtime_request.v2", requestId: "permission-1", type: "permission", - requestKind: "permission_approval", status: "pending", turnId: "turn-1", itemId: "write-1", + requestKind: "permission_approval", status: "pending", turnId: "durable-turn-1", itemId: "write-1", prompt: "Allow this write?", choices: [{ key: "deny", label: "Decline", outcome: "cancel" }], details: { toolCallId: "write-1" }, origin: { adapter: "acpx-runtime-sidecar", provider: "pi", method: "session/request_permission" }, @@ -27,10 +27,11 @@ function snapshot() { describe("live ACP request recovery", () => { it("retains original question, option and permission identities", () => { const value = snapshot(); - const requests = liveAcpxRuntimeRequests(value, "thread-1", "turn-1"); + const requests = liveAcpxRuntimeRequests(value, "thread-1", "turn-1", "durable-turn-1"); expect(requests.map(request => [request.id, request.method])).toEqual([ ["input-1", "elicitation/create"], ["permission-1", "session/request_permission"], ]); + expect(requests.every(request => request.params.turnId === "turn-1")).toBe(true); expect(requests[0]!.params.questionSet).toEqual(value.pendingRuntimeRequests[0]!.input); expect(requests[1]!.params).toMatchObject({ choices: value.pendingRuntimeRequests[1]!.choices, toolCallId: "write-1" }); }); @@ -39,10 +40,11 @@ describe("live ACP request recovery", () => { ["missing live attestation", { runtimeRequestsLive: false }], ["wrong provider", { provider: "codex" }], ["wrong session", { driverSessionId: "other" }], - ["wrong turn", { activeProviderTurnId: "other" }], + ["wrong provider turn", { activeProviderTurnId: "other" }], + ["wrong durable turn", { runtimeRequestTurnId: "other" }], ["missing ledger", { pendingRuntimeRequests: null }], ])("rejects %s", (_name, change) => { - expect(() => liveAcpxRuntimeRequests({ ...snapshot(), ...change }, "thread-1", "turn-1")).toThrow(); + expect(() => liveAcpxRuntimeRequests({ ...snapshot(), ...change }, "thread-1", "turn-1", "durable-turn-1")).toThrow(); }); it.each([ @@ -51,18 +53,18 @@ describe("live ACP request recovery", () => { { origin: { method: "item/tool/call" } }, { input: null }, ])("rejects malformed or stale callback %j", change => { const value = snapshot(); - expect(() => liveAcpxRuntimeRequests({ ...value, pendingRuntimeRequests: [{ ...value.pendingRuntimeRequests[0], ...change }] }, "thread-1", "turn-1")).toThrow(); + expect(() => liveAcpxRuntimeRequests({ ...value, pendingRuntimeRequests: [{ ...value.pendingRuntimeRequests[0], ...change }] }, "thread-1", "turn-1", "durable-turn-1")).toThrow(); }); it("rejects duplicate or oversized callback ledgers", () => { const value = snapshot(); for (const count of [2, 1025]) { - expect(() => liveAcpxRuntimeRequests({ ...value, pendingRuntimeRequests: Array(count).fill(value.pendingRuntimeRequests[0]) }, "thread-1", "turn-1")).toThrow(); + expect(() => liveAcpxRuntimeRequests({ ...value, pendingRuntimeRequests: Array(count).fill(value.pendingRuntimeRequests[0]) }, "thread-1", "turn-1", "durable-turn-1")).toThrow(); } }); it.each(["input-1", "permission-1"])("restores and delivers %s once after controller recovery", async requestId => { - const requests = liveAcpxRuntimeRequests(snapshot(), "thread-1", "turn-1"); + const requests = liveAcpxRuntimeRequests(snapshot(), "thread-1", "turn-1", "durable-turn-1"); const originalTransport = new FakeCodexTransport(); const original = await makeDriver([originalTransport]).openSession({ runId: "run-1", normalizedSessionId: "session-1", workingDirectory: WORKSPACE }); let recovered: Awaited> | undefined; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index c884fad4c60..b9742cc1103 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -945,9 +945,10 @@ function acpxPermissionRequestParams(request: Record, threadId: } /** Rehydrate only the pending ledger attested by the same live ACP session. */ -export function liveAcpxRuntimeRequests(snapshot: Record, threadId: string, turnId: string): CodexRpcServerRequest[] { +export function liveAcpxRuntimeRequests(snapshot: Record, threadId: string, turnId: string, durableTurnId: string): CodexRpcServerRequest[] { if (snapshot.runtimeRequestsLive !== true || snapshot.provider !== "acpx" || snapshot.driverSessionId !== threadId || snapshot.activeProviderTurnId !== turnId + || snapshot.runtimeRequestTurnId !== durableTurnId || !durableTurnId || !turnId || !Array.isArray(snapshot.pendingRuntimeRequests) || snapshot.pendingRuntimeRequests.length > 1024) { throw new Error("ACPX pending request snapshot binding is invalid"); @@ -958,7 +959,7 @@ export function liveAcpxRuntimeRequests(snapshot: Record, threa const id = request.requestId, method = origin.method; if (typeof id !== "string" || !id || id.length > 160 || ids.has(id) || request.schema !== "paperclip.runtime_request.v2" || request.status !== "pending" - || request.turnId !== turnId || typeof method !== "string") { + || request.turnId !== durableTurnId || typeof method !== "string") { throw new Error("ACPX pending request snapshot request is invalid"); } ids.add(id); @@ -3796,7 +3797,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } } if (restoreRuntimeRequests && activeProviderTurnId !== null) { - this.#restoredRuntimeRequests = liveAcpxRuntimeRequests(snapshot, this.#threadId, activeProviderTurnId); + this.#restoredRuntimeRequests = liveAcpxRuntimeRequests(snapshot, this.#threadId, activeProviderTurnId, this.#durableTurnId); for (const request of this.#restoredRuntimeRequests) { this.#bridgedRuntimeInputs.set(String(request.id), { durableTurnId: this.#durableTurnId, diff --git a/tests/runner-e2e/FIXTURES.md b/tests/runner-e2e/FIXTURES.md index c709ef353f3..4dd55988a01 100644 --- a/tests/runner-e2e/FIXTURES.md +++ b/tests/runner-e2e/FIXTURES.md @@ -397,6 +397,11 @@ macOS runs metadata negatives and explicitly skips this Linux-only calibration. That calibration and the earlier fake-Pi wrapper/bridge tests do not count as the real paid Product lifecycle proof. This new candidate cell remains unqualified. +The pending-question controller-restart browser matcher accepts only the retained +issue's UUID or public identifier and the exact retained interaction ID. The UI +normally posts with the public identifier. Durable request, run, turn, session, +producer and single-delivery assertions remain required after submission. + ## Pi active controls The explicit-only `pi-controls` suite adds `pending-permission-stop` and diff --git a/tests/runner-e2e/pi-native-restart-flow.test.ts b/tests/runner-e2e/pi-native-restart-flow.test.ts index 475b5b72c11..f9a69039543 100644 --- a/tests/runner-e2e/pi-native-restart-flow.test.ts +++ b/tests/runner-e2e/pi-native-restart-flow.test.ts @@ -164,16 +164,20 @@ it.each([null, {}, { status: "answered", value: "guess" }, { status: "cancelled" }, ); -async function flow(failure?: "replaced" | "wrong-file" | "missing-resolution" | "replaced-runner") { +async function flow(failure?: "replaced" | "wrong-file" | "missing-resolution" | "replaced-runner", issueRef = "issue") { const f = fixture(), evidence = new Map(), checkpoints: string[] = []; + Object.assign(f.state.issue, { identifier: "RUN-1" }); let answer = "", restarts = 0, submissions = 0, proof: unknown, resolvePost: ((value: unknown) => void) | undefined; let postPredicate: ((value: any) => boolean) | undefined; const composer: any = { count: () => 1, filter: () => composer, locator: () => composer, first: () => composer, fill: async (value: string) => { expect(restarts).toBe(1); answer = value; } }; const button: any = { count: () => 1, filter: () => button, click: async () => { expect(restarts).toBe(1); expect(answer).toMatch(/^PI-RESTART-[a-f0-9]{32}$/); submissions++; - const request = { url: () => "http://fixture/api/issues/issue/interactions/interaction/respond", method: () => "POST", + const request = { url: () => `http://fixture/api/issues/${issueRef}/interactions/interaction/respond`, method: () => "POST", postDataJSON: () => ({ answers: [{ questionId: "answer", optionIds: [], otherText: answer }] }) }; + expect(postPredicate!({ ...request, url: () => "http://fixture/api/issues/OTHER-1/interactions/interaction/respond" })).toBe(false); + expect(postPredicate!({ ...request, url: () => `http://fixture/api/issues/${issueRef}/interactions/other/respond` })).toBe(false); + expect(postPredicate!({ ...request, method: () => "GET" })).toBe(false); expect(postPredicate!(request)).toBe(true); resolvePost!(request); proof = f.finish(answer); if (failure === "missing-resolution") f.events.splice(1, 1); } }; @@ -201,5 +205,5 @@ async function flow(failure?: "replaced" | "wrong-file" | "missing-resolution" | expect(JSON.stringify(evidence.get("pi-native-restart-before.json"))).not.toContain("PI-RESTART-"); if (!["replaced", "replaced-runner"].includes(failure ?? "")) expect(evidence.has("pi-native-restart-final.json")).toBe(true); } -it("restarts while unanswered, reloads, then submits hidden text through the exact browser route", () => flow()); +it.each(["issue", "RUN-1"])("restarts while unanswered and submits through the exact %s browser route", issueRef => flow(undefined, issueRef)); it.each(["replaced", "wrong-file", "missing-resolution", "replaced-runner"] as const)("retains evidence and fails whole flow on %s", failure => flow(failure)); diff --git a/tests/runner-e2e/pi-native-restart-flow.ts b/tests/runner-e2e/pi-native-restart-flow.ts index b2488a0214e..34a34739d3d 100644 --- a/tests/runner-e2e/pi-native-restart-flow.ts +++ b/tests/runner-e2e/pi-native-restart-flow.ts @@ -161,8 +161,9 @@ export async function runPiPendingControllerRestart(input: { await expect(composer()).toHaveCount(1); await input.capture("pi-restart-reconnected", "Same Pi question after controller restart", "pi-restart-reconnected.png"); await composer().locator('[contenteditable="true"],textarea').first().fill(answer); - const route = `/api/issues/${pending.issueId}/interactions/${pending.interactionId}/respond`; - const submitted = input.page.waitForRequest(request => new URL(request.url()).pathname === route && request.method() === "POST", + const issueRefs = [pending.issueId, after.state.issue.identifier].filter(id); + const routes = new Set(issueRefs.map(issueRef => `/api/issues/${encodeURIComponent(issueRef)}/interactions/${pending.interactionId}/respond`)); + const submitted = input.page.waitForRequest(request => routes.has(new URL(request.url()).pathname) && request.method() === "POST", { timeout: Math.max(1, input.deadlineAt - Date.now()) }); const button = input.page.getByRole("button", { name: after.state.interactions[0]!.payload.questionSet.submitLabel ?? "Submit answers", exact: true }).filter({ visible: true }); await expect(button).toHaveCount(1); await button.click();