fix(runner): bind restored ACP requests to durable and provider turns

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-02 13:04:41 -05:00
1 parent 2b50801b24
commit 780471702c
6 files changed
+142 -38

No files matched your search

@@ -733,6 +733,33 @@ fn validate_pending_runtime_requests(
Ok(())
}
fn attested_pending_runtime_requests(
pending: &BTreeMap<String, Value>,
provider: &crate::acpx_provider_state::AcpxProviderState,
live: &BTreeMap<String, String>,
durable_turn_id: &str,
) -> Vec<Value> {
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::<Vec<_>>();
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();
@@ -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<ReturnType<typeof original.snapshot>> | undefined;
@@ -945,9 +945,10 @@ function acpxPermissionRequestParams(request: Record<string, unknown>, threadId:
}
/** Rehydrate only the pending ledger attested by the same live ACP session. */
export function liveAcpxRuntimeRequests(snapshot: Record<string, unknown>, threadId: string, turnId: string): CodexRpcServerRequest[] {
export function liveAcpxRuntimeRequests(snapshot: Record<string, unknown>, 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<string, unknown>, 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,
+5
View File
@@ -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
@@ -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<string, any>(), 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));
+3 -2
View File
@@ -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();