From 9d19f98b50d5e5f64d9d78ceff7e6049978cf048 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:35:38 -0500 Subject: [PATCH] fix: harden native chat recovery and add coordination evals (#13741) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip helps people manage AI agents and their work. > - Agent chat uses native runner sessions to plan, delegate, and track that work. > - A user can press Stop while the native session is still starting. > - The server can acknowledge that Stop without dispatching it, then let the session submit a turn. > - This leaves chat recovery waiting for an execution that the user expected to stop. > - This PR waits for the startup handle, dispatches cancellation, and prevents a late startup from submitting a turn. > - New full-stack evals check the resulting records and outputs across Claude and Codex. > - Those evals also exposed missing ACPX readiness fields, unbounded polling, and an old-run identity check that rejected valid warm handoffs. ## Linked Issues or Issue Description **What happened?** Stop during native startup could record an acknowledged cancellation with `dispatched: false`. The provider could then begin work. A subsequent `/new` stayed queued. A remote Claude follow-up also exhausted the command journal while probing warm-session readiness: ACPX never returned the readiness fields required by the shared transport. Once readiness worked, attachment incorrectly compared the next run descriptor against the old run ID. The 25 ms polling loop could issue 4,800 commands during its two-minute wait, beyond the 500-command bound. The existing chat eval treated lifecycle logs as proof of an active provider turn, so it did not distinguish startup cancellation from active-turn cancellation. **Expected behavior** A Stop during startup must reach the pending session. A late session must not submit a prompt after Stop. Recovery must retain control when startup exceeds the bounded wait. Chat evals must check saved task state, document contents, worker identity, account binding, and duplicate effects. **Steps to reproduce** 1. Start a native Claude or Codex chat turn. 2. Press Stop after process startup is requested but before the provider turn starts. 3. Send `/new`, then send a fresh message. 4. On the affected base, cancellation can be acknowledged without dispatch and the reset stays queued. **Paperclip version or commit** The live Claude baseline reproduced this on `29d6b3509`. The branch also includes master commit `0f5fafe16`. Related work: #13678, #13686, #13693, #13291, #13738. A separate runner reliability branch also contains a startup-wait fix. Its overlap must be reconciled before merging; this branch additionally prevents prompt submission after a late startup. ## What Changed - Wait for a pending native startup before acknowledging a run-scoped Stop. Preserve the existing recovery error when that wait expires. - Keep a Stop guard on startup. Cancel a late handle before it can submit a provider turn. - Add regression tests for normal handle publication and publication after the Stop deadline. - Back off blocked warm-attachment probes. Keep the fast two-snapshot barrier, fail closed, and record changed blockers. - Add red/green tests for delayed readiness, persistent blockers, alternating readiness, and readiness near the deadline. - Publish ACPX readiness and blockers. Preserve the old authority’s event acknowledgement barrier; only settled sessions can proceed to attachment. - Bind warm ACPX descriptors to the validated next authority while retaining old-run event correlation until activation. Preserve session identity and provider profile checks. - Exercise two consecutive run rotations through a qualified fake sidecar, verifying checkpointing, provider identity, pre-activation rejection, and new-run work admission. - Separate startup and active-turn cancellation checkpoints in the browser eval. - Add 18 explicit native chat eval cells: 12 local and 6 Daytona cells across Claude and Codex. - Cover hiring and reuse through managed AI accounts, source-based review, current blocked-task status, request replay after a lost HTTP acknowledgement, server restart continuity, and Stop/reset continuity. - Use ordinary production agent instructions. Enable API tools only for the two coordination cases that need them. - Calibrate the matchers with invalid records and outputs. Require remembered context after restart and a structured status snapshot that distinguishes the current blocker from history and task status from active execution. Compare the public issue mutation contract and relationships during read-only reporting. Preserve before/after source records in failed eval evidence. - Fix the lost-ack browser harness and verify it against a real HTTP server. Check the chat composer after restart instead of waiting for an unrelated document lifecycle event. - Document the scope and limits of each case. ## Verification - The startup regression failed on the unfixed executor and passed after the fix. - `pnpm test:e2e:runner:typecheck` passed. - `pnpm test:e2e:runner:unit` passed: 424 tests in 37 files. - `pnpm exec vitest run server/src/services/native-runtime/native-session-executor.test.ts` passed: 385 tests. - [Baseline live campaign](https://github.com/paperclipai/paperclip/actions/runs/35608208868): Claude Stop reproduced the bug. Codex Stop and Claude hire/reuse passed. Codex delegation was blocked by provider capacity. - [Eval-only startup campaign](https://github.com/paperclipai/paperclip/actions/runs/35609479786): both providers failed as expected. Both persisted `dispatched: false` and left `/new` queued. - [First fixed campaign](https://github.com/paperclipai/paperclip/actions/runs/35610533706) on `c9e95797d`: 10/18 cells passed. Startup Stop passed for both providers. Failed cases exposed eval harness defects and remote continuity failures. All attempts remain available. - [Original workflows and stronger memory checks](https://github.com/paperclipai/paperclip/actions/runs/35611896649) on `c04324fab`: 9/12 passed. Reassignment, local restart memory, and startup Stop passed for both providers; Codex remote restart passed. Claude remote restart exposed the missing readiness contract. Two Codex planning cells hit provider capacity. - [Unchanged-model retry](https://github.com/paperclipai/paperclip/actions/runs/35613854548): Codex planning and backlog creation both passed. - [18-cell campaign with ACPX readiness](https://github.com/paperclipai/paperclip/actions/runs/35614586963) on `6a98ef743`: 16/18 passed, including all local/remote Stop and committed-send cases. Claude remote continuity exposed the next-authority check, now fixed. Codex hiring produced its checklist, but the runner redacted the requested marker after it appeared as “Tracking token: …”. That content-redaction policy is unchanged and remains an explicit limitation. - [Structured status grading](https://github.com/paperclipai/paperclip/actions/runs/35614954725) on `50448c228`: both providers passed on their first attempt, including cleanup. - [Complete read-only state grading](https://github.com/paperclipai/paperclip/actions/runs/35616089011) on `551e13892`: both providers passed. - [Final ACPX handoff and hiring retry](https://github.com/paperclipai/paperclip/actions/runs/35617045456) on `cbd637587`: all three Claude Daytona cases passed (restart continuity, active Stop/reset, and lost-ack replay). Codex hiring reproduced the content-redaction failure: the saved checklist contained `Tracking token: [REDACTED]` instead of the required business marker. All four cases completed cleanup successfully. [Published report](https://d1p6rlowie26tp.cloudfront.net/runner-e2e/campaigns/gha-35617045456-1/). The only subsequent commit adds the qualified-sidecar integration test; production code is identical to this live proof. - `pnpm test:e2e:runner:browser-support` passed: 5 browser tests without paid models. - Runner TypeScript typecheck passed. All 5 warm-readiness tests pass; two failed with the prior fixed-rate loop, and the late-readiness test failed before the pacing correction. - ACPX readiness and warm-identity regressions each failed before their fixes. All 292 runner-core Rust library tests passed. The qualified-sidecar integration test passes. Rust formatting is checked. - Status-grader regressions for misleading historical mentions and previously unchecked mutations each failed before tightening the oracle and pass now. - [Latest-head CI](https://github.com/paperclipai/paperclip/actions/runs/35617522307) passed on `a4093c8f1`: full build, type checks, test partitions, browser E2E, and native runner checks. Two unrelated tests initially failed (Sentry fixture release attribution and local-service fixture readiness); both passed locally together (35 passed, 5 optional SDK tests skipped) and on the failed-job retry. No changes were made to those tests. - Greptile reviewed `a4093c8f1` at 5/5; both earlier findings are fixed and all review threads are resolved. - The paid live suite is not fully green: the reproducible content-redaction case remains red. This is separate from the passing PR merge checks. No production content-redaction, prompt, model, or completion-policy change is included. - Managed-account hiring and review cases explicitly enable API tools; these do not qualify default new-user onboarding. ## Risks - Stop can wait up to 30 seconds for startup, then use the existing pending-recovery path. This does not prove that remote cleanup has finished. - Blocked warm readiness adds up to 750 ms between later probes with the two-minute remote budget, or about 32 ms with the default five-second budget. Ready sessions retain the short second barrier. - Paid evals can fail because of provider capacity or agent decisions. Each failure needs evidence-based classification. - The HTTP request replay case checks comment idempotency and duplicate effects. It does not prove replay safety for an ambiguous provider tool call. - The new suite is opt-in. It does not increase the default paid campaign. - No production prompts or model selection change. Review-handoff behavior and content-redaction policy remain separate product decisions. The latter can remove harmless business content that looks like credential syntax; the failing attempt is retained. ## Model Used OpenAI Codex, GPT-6, with repository tools and code execution. The exact deployment model ID and context window are not exposed in this task. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/evals.md | 6 + doc/execution-semantics.md | 24 + .../runner-core/src/acpx_provider_backend.rs | 142 +++++- .../runner-core/tests/acpx_warm_attachment.rs | 166 +++++++ .../src/live/runnerd-codex-transport.ts | 31 +- .../live/warm-attachment-readiness.test.ts | 66 +++ .../src/live/warm-attachment-readiness.ts | 36 ++ .../native-session-executor.test.ts | 95 ++++ .../native-runtime/native-session-executor.ts | 431 ++++++++++-------- tests/runner-e2e/FIXTURES.md | 19 +- tests/runner-e2e/README.md | 53 ++- tests/runner-e2e/catalog.test.ts | 6 +- tests/runner-e2e/catalog.ts | 15 +- tests/runner-e2e/chat-cases.ts | 23 +- tests/runner-e2e/chat-flow.ts | 46 +- tests/runner-e2e/chat-hardening.test.ts | 117 +++++ tests/runner-e2e/chat-hardening.ts | 244 ++++++++++ tests/runner-e2e/harness-env.ts | 3 +- tests/runner-e2e/live-fixtures.test.ts | 15 +- tests/runner-e2e/live-fixtures.ts | 5 +- tests/runner-e2e/lost-send.spec.ts | 50 ++ tests/runner-e2e/lost-send.ts | 37 ++ tests/runner-e2e/playwright-support.config.ts | 2 +- 23 files changed, 1373 insertions(+), 259 deletions(-) create mode 100644 packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs create mode 100644 packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts create mode 100644 packages/paperclip-runner/src/live/warm-attachment-readiness.ts create mode 100644 tests/runner-e2e/chat-hardening.test.ts create mode 100644 tests/runner-e2e/chat-hardening.ts create mode 100644 tests/runner-e2e/lost-send.spec.ts create mode 100644 tests/runner-e2e/lost-send.ts diff --git a/doc/evals.md b/doc/evals.md index 88fedec15ed..842422bf1c7 100644 --- a/doc/evals.md +++ b/doc/evals.md @@ -60,6 +60,12 @@ harness are under `tests/runner-e2e`; the package scripts are `test:e2e:runner`, the server and embedded database, creates the company/agent/task through the real APIs, drives Chromium, and invokes the selected local or Daytona runner. +The explicit-only `agent-chat-hardening` Product E2E suite covers native chat +recovery, hiring, status evidence, and review handoff on local and selected warm +Daytona paths. Its [fixture contract](../tests/runner-e2e/README.md) distinguishes +startup cancellation from active response cancellation and HTTP send replay +from ambiguous provider action recovery. Select it explicitly; `--all` excludes it. + ## Validation ladder Start with credential-free checks and a catalog listing. For Product E2E: diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 5c2cae81a3e..bfaee4b8443 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1093,6 +1093,15 @@ controller lease in the same transaction that creates the native coordinator. The native executor rechecks cancellation and terminal status when claiming the coordinator, before starting or attaching a provider. +Run-only Stop also covers the interval after the coordinator claim and before +the provider session publishes its handle. Stop retains its pending audited +intent and waits up to 30 seconds for that startup to settle. A published +session receives cancellation before prompt submission; only real dispatch +sets `dispatched: true`. A deadline leaves the intent pending and the late +session remains fenced and is closed. Stop acknowledgement alone does not +certify cleanup: the existing process and environment receipts still govern +admission of the next message. + A cancelled startup can continue from a newer authenticated user message after cleanup. The server requires either its explicit before-selection fence or an unclaimed native coordinator (zero attempts and controller generations, no @@ -1212,6 +1221,21 @@ session. A staged provider package is reused only after the complete expected manifest and artifact hashes verify. A missing, changed, or incompatible package must be replaced and verified before launch. +Warm attachment requires two consecutive authenticated readiness snapshots. +Blocked readiness probes back off within the reconnect deadline so they do not +fill the durable command journal while waiting. The fast ready path keeps its +short second barrier. If readiness never arrives, attachment fails closed with +the last observed blocker; a full journal is not a substitute for that diagnosis. +Both native providers publish this readiness contract. ACPX reports its durable +session identity, active turn, pending audit events, closed state, and unproven +provider exit as blockers. Explicit readiness probes let the durable runner +commit and acknowledge retained events under the old run authority; snapshotting +alone never discards them. ACPX checkpoints its process during the subsequent +attachment before resuming the same provider session under the new run. +During an in-place handoff, the ACPX descriptor binds to the validated next run +while event correlation stays on the old run until durable authority activation. +A changed session identity or a descriptor that names any other run is rejected. + Safe native replacement may clear a Blocked status only with a durable receipt that the same failed run projected that exact status version. Explicitly reasserting Blocked or changing its blockers advances the status version, even 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 6805294ec00..e4cf725f1fe 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 @@ -799,7 +799,32 @@ impl AcpxCommandExecutor { .map_err(|error| { DurableRunnerError::invalid(format!("run.attach ACPX provider is invalid: {error}")) })?; - descriptor.validate(&self.context)?; + let mut attachment_context = self.context.clone(); + if let Some(boundary) = payload.get("paperclipNextAuthority") { + // The durable runner validates this authority's immutable bindings + // and PRP v2 handoff capability before dispatching run.attach. Its + // activation follows our command result, so self.context must keep + // correlating audit events with the old run until rotate_authority. + let identity = boundary.get("identity").ok_or_else(|| { + DurableRunnerError::invalid("run.attach authority identity is required") + })?; + let run_id = identity + .get("runId") + .and_then(Value::as_str) + .ok_or_else(|| { + DurableRunnerError::invalid("run.attach authority runId is required") + })?; + if run_id == self.context.run_id + || identity.get("normalizedSessionId").and_then(Value::as_str) + != Some(self.context.normalized_session_id.as_str()) + { + return Err(DurableRunnerError::invalid( + "run.attach authority changed an immutable session binding", + )); + } + attachment_context.run_id = run_id.to_owned(); + } + descriptor.validate(&attachment_context)?; let tool_set = authorized_tool_set(payload)?; let state = self .state @@ -1260,6 +1285,27 @@ impl AcpxCommandExecutor { .state .as_ref() .ok_or_else(|| DurableRunnerError::invalid("ACPX provider has not been prepared"))?; + // The shared transport waits for an explicit readiness contract before + // rotating run authority. ACPX checkpoints its process during attach, + // but must first settle the same durable guards enforced by attach_run. + // Do not consume events here: the durable runner commits and ACKs the + // old authority's retained prefix before a later probe can report ready. + let mut warm_attach_blockers = Vec::new(); + if state.lifecycle == "closed" { + warm_attach_blockers.push("durable_closed"); + } + if state.provider_exit_unconfirmed { + warm_attach_blockers.push("provider_exit_unconfirmed"); + } + if state.identity.is_none() { + warm_attach_blockers.push("provider_identity_unavailable"); + } + if state.active_turn_id.is_some() { + warm_attach_blockers.push("durable_active_turn"); + } + if !state.pending_events.is_empty() { + warm_attach_blockers.push("durable_pending_events"); + } Ok(CommandExecution::result(json!({ "status": state.lifecycle, "provider": "acpx", @@ -1270,6 +1316,8 @@ impl AcpxCommandExecutor { "providerAccountSessionId": state.identity.as_ref().map(|value| value.agent_session_id.as_str()), "providerIdentity": state.identity, "activeProviderTurnId": state.active_turn_id, + "warmAttachReady": warm_attach_blockers.is_empty(), + "warmAttachBlockers": warm_attach_blockers, }))) } @@ -2099,6 +2147,61 @@ mod tests { let original_config = test_config(&directory, Some(launch_profile.clone())); let mut original = AcpxCommandExecutor::with_runner_config(&directory, &original_config); original.state = Some(state); + let settled = original.state.clone().unwrap(); + let ready = original.snapshot().unwrap().result; + assert_eq!(ready["warmAttachReady"], true); + assert_eq!(ready["warmAttachBlockers"], json!([])); + let blocked_states: [(&str, fn(&mut AcpxDurableState)); 4] = [ + ("durable_closed", |state| { + state.lifecycle = "closed".to_owned() + }), + ("provider_exit_unconfirmed", |state| { + state.lifecycle = "prepared".to_owned(); + state.provider_exit_unconfirmed = true; + }), + ("provider_identity_unavailable", |state| { + state.lifecycle = "prepared".to_owned(); + state.identity = None; + }), + ("durable_active_turn", |state| { + state.lifecycle = "turn_active".to_owned(); + state.active_turn_id = Some("turn-1".to_owned()); + }), + ]; + for (blocker, mutate) in blocked_states { + original.state = Some(settled.clone()); + mutate(original.state.as_mut().unwrap()); + let snapshot = original.snapshot().unwrap().result; + assert_eq!(snapshot["warmAttachReady"], false, "{blocker}"); + assert_eq!(snapshot["warmAttachBlockers"], json!([blocker])); + } + // A readiness probe must retain the old authority's audit events until + // the durable runner commits and acknowledges them, including recovery + // notices that run.attach itself is allowed to consume. + for event_type in ["session.resumed", "harness.diagnostic"] { + original.state = Some(settled.clone()); + original + .state + .as_mut() + .unwrap() + .push(NormalizedProviderEvent { + event_type: event_type.to_owned(), + priority: EventPriority::P0, + payload: json!({}), + }) + .unwrap(); + let snapshot = original.snapshot().unwrap().result; + assert_eq!(snapshot["warmAttachReady"], false); + assert_eq!( + snapshot["warmAttachBlockers"], + json!(["durable_pending_events"]) + ); + assert_eq!(original.retained_events().unwrap().len(), 1); + original.acknowledge_events(1).unwrap(); + assert_eq!(original.snapshot().unwrap().result["warmAttachReady"], true); + } + original.state = Some(settled); + assert!(!marker.exists(), "readiness must not start a provider"); original.save_state().unwrap(); let mut wrong_session_config = original_config.clone(); @@ -2135,6 +2238,43 @@ mod tests { .unwrap(); assert_eq!(attached.state.as_ref().unwrap().descriptor.run_id, "run-2"); assert!(!marker.exists()); + + // In-place warm handoff executes under the old authority. Only the + // authenticated next-authority boundary may admit the new descriptor; + // event correlation stays on run-1 until durable activation completes. + assert!(original + .attach_run(&json!({"provider": descriptor_value})) + .is_err()); + let warm_payload = json!({ + "provider": descriptor_value, + "paperclipNextAuthority": { + "identity": { + "runnerInstanceId": original_config.runner_instance_id, + "environmentLeaseId": original_config.environment_lease_id, + "runId": "run-2", + "normalizedSessionId": original_config.normalized_session_id, + "turnId": "turn-2", + "itemId": "item-2", + }, + "connection": {"mode": "connect", "connectUrl": original_config.connect_url}, + }, + }); + let mut wrong_run = warm_payload.clone(); + wrong_run["paperclipNextAuthority"]["identity"]["runId"] = json!("run-3"); + assert!(original.attach_run(&wrong_run).is_err()); + let mut wrong_session = warm_payload.clone(); + wrong_session["paperclipNextAuthority"]["identity"]["normalizedSessionId"] = + json!("other-session"); + assert!(original.attach_run(&wrong_session).is_err()); + let mut changed_profile = warm_payload.clone(); + changed_profile["provider"]["instructions"] = json!("different profile"); + assert!(original.attach_run(&changed_profile).is_err()); + original.attach_run(&warm_payload).unwrap(); + assert_eq!(original.state.as_ref().unwrap().descriptor.run_id, "run-2"); + assert_eq!(original.context.run_id, "run-1"); + original.rotate_authority(&attached_config); + assert_eq!(original.context.run_id, "run-2"); + assert!(!marker.exists()); fs::remove_dir_all(directory).unwrap(); } diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs new file mode 100644 index 00000000000..d9673780217 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs @@ -0,0 +1,166 @@ +#![cfg(unix)] + +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::path::PathBuf; +use std::time::Duration; + +use paperclip_runner_core::acpx_provider_backend::AcpxCommandExecutor; +use paperclip_runner_core::durable::{ + AcpxLaunchProfile, Command, CommandExecutor, DurableRunnerConfig, QualifiedLaunchArtifact, +}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +const PROFILE_DIGEST: &str = + "sha256:c4538599d1ab767db5dff50934f13bb5ba313a59d9c4a83e993fac4617ea63d3"; + +struct Fixture(PathBuf); +impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +fn command(sequence: u64, kind: &str, payload: Value) -> Command { + Command { + schema: "paperclip.prp.command.v1".to_owned(), + command_id: format!("command-{sequence}"), + controller_seq: sequence, + command_type: kind.to_owned(), + issued_at: "2026-09-21T00:00:00.000Z".to_owned(), + deadline_at: None, + precondition: None, + payload, + } +} + +#[test] +fn checkpoints_the_sidecar_and_rebinds_consecutive_warm_runs_before_accepting_work() { + let fixture = Fixture(std::env::temp_dir().join(format!("acpx-warm-{}", uuid::Uuid::new_v4()))); + fs::create_dir_all(&fixture.0).unwrap(); + fs::set_permissions(&fixture.0, fs::Permissions::from_mode(0o700)).unwrap(); + let sidecar = fixture.0.join("sidecar"); + fs::copy(env!("CARGO_BIN_EXE_fake-acpx-sidecar"), &sidecar).unwrap(); + fs::set_permissions(&sidecar, fs::Permissions::from_mode(0o700)).unwrap(); + let args = vec![ + "--mode".to_owned(), + "suspend".to_owned(), + "--profile-digest".to_owned(), + PROFILE_DIGEST.to_owned(), + ]; + let mut config = DurableRunnerConfig { + connect_url: "ws://127.0.0.1/runner".to_owned(), + ca_bundle_path: None, + state_dir: fixture.0.clone(), + runner_instance_id: "runner-1".to_owned(), + environment_lease_id: "lease-1".to_owned(), + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + runner_version: "0.0.0".to_owned(), + runner_digest: format!("sha256:{}", "a".repeat(64)), + acpx_launch_profile: Some(AcpxLaunchProfile { + authority_digest: format!("sha256:{}", "d".repeat(64)), + command: sidecar.clone(), + args: args.clone(), + artifacts: vec![QualifiedLaunchArtifact { + path: sidecar.clone(), + sha256: format!("sha256:{:x}", Sha256::digest(fs::read(&sidecar).unwrap())), + }], + }), + opencode_launch_profile: None, + max_outbox_bytes: 1024 * 1024, + p0_reserve_bytes: 64 * 1024, + max_frame_bytes: 1024 * 1024, + reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, + max_runtime: Duration::from_secs(30), + }; + let mut descriptor = json!({ + "kind": "acpx", "provider": "acpx", "driver": "acpx_runtime", + "providerVersion": "0.13.1", "agent": "codex", "model": "gpt-5.6-sol", + "acpxVersion": "0.13.1", "agentServerPackage": "@agentclientprotocol/codex-acp", + "agentServerVersion": "1.6.2", "agentRuntimePackage": "@openai/codex", + "agentRuntimeVersion": "0.153.4", "commandDigest": PROFILE_DIGEST, + "sidecarCommand": sidecar, "sidecarArgs": args, "runtimeDirectory": fixture.0, + "normalizedSessionId": "session-1", "runId": "run-1", "cwd": fixture.0, + "instructions": "Complete the supplied work.", + "permissionMode": "approve-all", "permissionModePinned": true, + }); + let mut executor = AcpxCommandExecutor::with_runner_config(&fixture.0, &config); + executor + .execute(&command(1, "run.prepare", json!({"provider": descriptor}))) + .unwrap(); + let opened = executor + .execute(&command(2, "session.open", json!({}))) + .unwrap(); + let provider_id = opened.result["providerSessionId"].clone(); + let mut previous_pid = opened.result["processId"].clone(); + assert!(previous_pid.is_number()); + + for run in 2..=3 { + let ready = executor + .execute(&command( + run * 10, + "session.snapshot", + json!({"quiesceForWarmAttach": true}), + )) + .unwrap(); + assert_eq!(ready.result["warmAttachReady"], true); + config.run_id = format!("run-{run}"); + config.turn_id = format!("turn-{run}"); + config.item_id = format!("item-{run}"); + descriptor["runId"] = json!(config.run_id); + let attached = executor + .execute(&command( + run * 10 + 1, + "run.attach", + json!({ + "provider": descriptor, + "paperclipNextAuthority": { + "identity": { + "runnerInstanceId": config.runner_instance_id, + "environmentLeaseId": config.environment_lease_id, + "runId": config.run_id, + "normalizedSessionId": config.normalized_session_id, + "turnId": config.turn_id, + "itemId": config.item_id, + }, + "connection": {"mode": "connect", "connectUrl": config.connect_url}, + }, + }), + )) + .unwrap(); + assert_eq!(attached.result["status"], "resumed"); + assert_eq!(attached.result["providerSessionId"], provider_id); + assert_ne!(attached.result["processId"], previous_pid); + previous_pid = attached.result["processId"].clone(); + assert!(attached + .events + .iter() + .any(|event| event.0 == "run.attached")); + // Provider preparation cannot grant the new run authority by itself. + assert!(executor + .execute(&command(run * 10 + 2, "session.snapshot", json!({}))) + .is_err()); + executor.rotate_authority(&config); + let active = executor + .execute(&command(run * 10 + 3, "session.snapshot", json!({}))) + .unwrap(); + assert_eq!(active.result["warmAttachReady"], true); + assert_eq!(active.result["providerSessionId"], provider_id); + } + let started = executor + .execute(&command( + 40, + "turn.start", + json!({ + "turnId": "provider-turn-3", "text": "Perform work under run-3.", + }), + )) + .unwrap(); + assert_eq!(started.result["status"], "accepted"); + executor.shutdown().unwrap(); +} diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index c244952fdfc..4b624890030 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1,4 +1,5 @@ import { RunnerdTraceFrameIndex } from "./runnerd-trace-frame-index.js"; +import { waitForWarmAttachmentReadiness } from "./warm-attachment-readiness.js"; import { codexExecutableReadOnlyRoots } from "../drivers/codex/codex-security-config.js"; import { isCanonicalProviderEventType } from "../provider-events.js"; import { execFileSync } from "node:child_process"; @@ -3670,34 +3671,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // same budget here so a transient tunnel reconnect cannot trip the shorter // generic command timeout and replace an otherwise healthy warm runner. const reconnectGraceMs = this.options.runnerReconnectGraceMs ?? 5_000; - const deadline = Date.now() + reconnectGraceMs; - let consecutiveReadyProbes = 0; - let lastBlockers: unknown = null; - while (Date.now() < deadline) { - await this.#awaitWarmRunnerConnection(deadline); - const snapshot = await this.#commandResult( + await waitForWarmAttachmentReadiness({ + graceMs: reconnectGraceMs, + waitForConnection: (deadline) => this.#awaitWarmRunnerConnection(deadline), + snapshot: (deadline) => this.#commandResult( "session.snapshot", { quiesceForWarmAttach: true, }, deadline, - ); - lastBlockers = snapshot.warmAttachBlockers; - if (snapshot.warmAttachReady === true) { - consecutiveReadyProbes += 1; - // A second barrier prevents a provider frame emitted immediately after - // its terminal notification from racing the authority rotation. Each - // snapshot wakes runnerd, polls the provider, and drains the preceding - // durable event prefix before the next probe. - if (consecutiveReadyProbes >= 2) return; - } else { - consecutiveReadyProbes = 0; - } - await new Promise((resolveWait) => setTimeout(resolveWait, 25)); - } - throw new Error( - `native_runner_warm_attachment_not_quiescent: ${JSON.stringify(lastBlockers)}`, - ); + ), + onBlocked: (blockers) => this.#diagnostic(`warm attachment awaiting quiescence: ${JSON.stringify(blockers)}`), + }); } async #awaitWarmRunnerConnection(deadline: number): Promise { diff --git a/packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts b/packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts new file mode 100644 index 00000000000..e6ecd484c35 --- /dev/null +++ b/packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts @@ -0,0 +1,66 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { waitForWarmAttachmentReadiness } from "./warm-attachment-readiness.js"; + +afterEach(() => vi.useRealTimers()); + +it("preserves the command journal while a remote provider takes a minute to settle", async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + let commands = 12; + const snapshot = vi.fn(async () => { + if (++commands >= 500) throw new Error("Durable PRP command journal bound exceeded."); + return { warmAttachReady: Date.now() >= 60_000, warmAttachBlockers: Date.now() >= 60_000 ? [] : ["pending_messages"] }; + }); + const result = waitForWarmAttachmentReadiness({ graceMs: 120_000, waitForConnection: async () => {}, snapshot }) + .then(() => "ready", error => String(error)); + await vi.advanceTimersByTimeAsync(120_000); + expect(await result).toBe("ready"); + expect(commands).toBeLessThan(200); +}); + +it("keeps the fast path and requires two consecutive ready snapshots", async () => { + vi.useFakeTimers(); + const sequence = [true, false, true, true]; + const snapshot = vi.fn(async () => ({ warmAttachReady: sequence.shift(), warmAttachBlockers: [] })); + const waiting = waitForWarmAttachmentReadiness({ graceMs: 5_000, waitForConnection: async () => {}, snapshot }); + await vi.advanceTimersByTimeAsync(200); + await waiting; + expect(snapshot).toHaveBeenCalledTimes(4); +}); + +it("fails closed with the actual blocker when quiescence never arrives", async () => { + vi.useFakeTimers(); + const snapshot = vi.fn(async () => ({ warmAttachReady: false, warmAttachBlockers: ["pending_runtime_requests"] })); + const result = waitForWarmAttachmentReadiness({ graceMs: 120_000, waitForConnection: async () => {}, snapshot }) + .then(() => "unexpected-ready", error => String(error)); + await vi.advanceTimersByTimeAsync(122_000); + expect(await result).toContain('native_runner_warm_attachment_not_quiescent: ["pending_runtime_requests"]'); + expect(snapshot.mock.calls.length).toBeLessThan(200); +}); + +it("does not reset the probe budget when readiness alternates", async () => { + vi.useFakeTimers(); + let calls = 0; + const snapshot = vi.fn(async () => ({ warmAttachReady: ++calls % 2 === 0, warmAttachBlockers: ["pending_events"] })); + const result = waitForWarmAttachmentReadiness({ graceMs: 120_000, waitForConnection: async () => {}, snapshot }) + .then(() => "unexpected-ready", error => String(error)); + await vi.advanceTimersByTimeAsync(122_000); + expect(await result).toContain("native_runner_warm_attachment_not_quiescent"); + expect(calls).toBeLessThan(350); +}); + +it("confirms readiness near the deadline after one unsettled observation", async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + let recoveredProbes = 0; + const result = waitForWarmAttachmentReadiness({ + graceMs: 5_000, + waitForConnection: async () => {}, + snapshot: async () => ({ + warmAttachReady: Date.now() >= 4_400 && ++recoveredProbes !== 2, + warmAttachBlockers: [], + }), + }).then(() => "ready", error => String(error)); + await vi.advanceTimersByTimeAsync(5_000); + expect(await result).toBe("ready"); +}); diff --git a/packages/paperclip-runner/src/live/warm-attachment-readiness.ts b/packages/paperclip-runner/src/live/warm-attachment-readiness.ts new file mode 100644 index 00000000000..70057c776a4 --- /dev/null +++ b/packages/paperclip-runner/src/live/warm-attachment-readiness.ts @@ -0,0 +1,36 @@ +/** Require two fresh authenticated barriers before rotating a warm run authority. */ +export async function waitForWarmAttachmentReadiness(input: { + graceMs: number; + waitForConnection: (deadline: number) => Promise; + snapshot: (deadline: number) => Promise>; + onBlocked?: (blockers: unknown) => void; +}): Promise { + const deadline = Date.now() + input.graceMs; + let consecutiveReadyProbes = 0; + let lastBlockers: unknown = null; + let probes = 0; + // Each probe is a durable command. A 25 ms loop over the remote 120 second + // reconnect budget can exhaust its 500-command journal before that budget. + // Leave room for both ready-confirmation probes and ordinary commands. The + // interval is proportional to the deadline, so a short local deadline never + // inherits a one-second delay after a transient readiness regression. + const blockedDelayMs = Math.max(25, Math.ceil(input.graceMs / 160)); + while (Date.now() < deadline) { + await input.waitForConnection(deadline); + const snapshot = await input.snapshot(deadline); + probes += 1; + if (snapshot.warmAttachReady !== true && JSON.stringify(snapshot.warmAttachBlockers) !== JSON.stringify(lastBlockers)) { + input.onBlocked?.(snapshot.warmAttachBlockers); + } + lastBlockers = snapshot.warmAttachBlockers; + if (snapshot.warmAttachReady === true) { + consecutiveReadyProbes += 1; + if (consecutiveReadyProbes >= 2) return; + } else { + consecutiveReadyProbes = 0; + } + const delayMs = snapshot.warmAttachReady === true || probes < 8 ? 25 : blockedDelayMs; + await new Promise(resolve => setTimeout(resolve, Math.min(delayMs, Math.max(0, deadline - Date.now())))); + } + throw new Error(`native_runner_warm_attachment_not_quiescent: ${JSON.stringify(lastBlockers)}`); +} diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index e244964226d..b8e49a833ae 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -4903,6 +4903,101 @@ describe("native session cancellation", () => { }); }); + it("waits for an in-flight startup handle before acknowledging Stop", async () => { + const root = await mkdtemp(join(tmpdir(), "native-startup-stop-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = root; + let open!: () => void, started!: () => void, finish!: () => void; + const opening = new Promise(resolve => { open = resolve; }); + const admitted = new Promise(resolve => { started = resolve; }); + const finished = new Promise(resolve => { finish = resolve; }); + const submitTurn = vi.fn(); + state.execute.mockReset().mockImplementationOnce(async (options) => { + started(); + await opening; + await options.onSession({ cancel: state.cancel }); + submitTurn(); + await finished; + await options.onSession(null); + return { + result: { summary: "cancelled" }, terminal: { runTerminalState: "cancelled" }, + turnId: "turn", normalizedSessionId: "session", providerSessionId: null, + driverKind: "test", driverVersion: "1", nativeEventCount: 0, + highestContiguousSourceSeq: 0, + }; + }); + const running = executePaperclipNativeSession({ + db: leaseDb(), execution, runnerInstanceId: "runner", useRunnerd: true, + }); + const outcome = running.catch(error => error); + const persistence = cancellationDb(); + let stopping: ReturnType | undefined; + let acknowledged = false; + try { + await admitted; + stopping = cancelNativeSession(execution.binding.runId, "operator Stop during startup", { + db: persistence.db, scope: "run", + }); + void stopping.then(() => { acknowledged = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(acknowledged).toBe(false); + expect(persistence.getResultJson().nativeCancellation).toMatchObject({ dispatchState: "pending" }); + open(); + await expect(stopping).resolves.toMatchObject({ dispatched: true }); + expect(state.cancel).toHaveBeenCalledOnce(); + expect(persistence.getResultJson().nativeCancellation).toMatchObject({ + dispatchState: "acknowledged", dispatched: true, + }); + finish(); + await outcome; + expect(submitTurn).not.toHaveBeenCalled(); + } finally { + open(); finish(); + await outcome; + await stopping; + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(root, { recursive: true, force: true }); + } + }); + + it("fences a late startup even when Stop reaches its acknowledgement deadline", async () => { + const root = await mkdtemp(join(tmpdir(), "native-late-startup-stop-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = root; + let open!: () => void, started!: () => void; + const opening = new Promise(resolve => { open = resolve; }); + const admitted = new Promise(resolve => { started = resolve; }); + const submitTurn = vi.fn(); + state.execute.mockReset().mockImplementationOnce(async (options) => { + started(); await opening; + await options.onSession({ cancel: state.cancel }); + submitTurn(); + throw new Error("a stopped startup must not reach prompt submission"); + }); + const outcome = executePaperclipNativeSession({ db: leaseDb(), execution, runnerInstanceId: "runner", useRunnerd: true }).catch(error => error); + const persistence = cancellationDb(); + try { + await admitted; + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const stopping = cancelNativeSession(execution.binding.runId, "operator Stop", { db: persistence.db, scope: "run" }).catch(error => error); + // Let the durable intent commit before advancing the startup deadline. + await new Promise(resolve => setImmediate(resolve)); + await vi.advanceTimersByTimeAsync(30_000); + expect(await stopping).toMatchObject({ name: "NativeCancellationPendingRecoveryError" }); + expect(persistence.getResultJson().nativeCancellation).toMatchObject({ dispatchState: "pending" }); + vi.useRealTimers(); + open(); await outcome; + expect(state.cancel).toHaveBeenCalledOnce(); + expect(submitTurn).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); open(); await outcome; + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(root, { recursive: true, force: true }); + } + }); + it("routes control-plane cancellation to the active normalized session and removes the handle", async () => { const running = executePaperclipNativeSession({ db: leaseDb(), diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 5abfbe1b55f..904b3a6073c 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -209,6 +209,8 @@ const nativeRunsDetachingForRestart = new Set(); type NativeSessionStartup = { promise: Promise; resolve: (session: ActiveNativeSession | null) => void; + stopRequested?: boolean; + cancellationSettled?: Promise; }; const nativeSessionStartups = new Map(); @@ -217,7 +219,10 @@ function detachActiveNativeSessionForRestart(active: ActiveNativeSession) { return active.restartDetach; } -async function waitForNativeStartupForRestart(runId: string) { +async function waitForNativeSessionStartup( + runId: string, + timeoutError: () => Error = () => new Error("native_restart_startup_not_ready"), +) { const startup = nativeSessionStartups.get(runId); if (!startup) return null; let timer: ReturnType | undefined; @@ -225,7 +230,7 @@ async function waitForNativeStartupForRestart(runId: string) { return await Promise.race([ startup.promise, new Promise((_resolve, reject) => { - timer = setTimeout(() => reject(new Error("native_restart_startup_not_ready")), 30_000); + timer = setTimeout(() => reject(timeoutError()), 30_000); }), ]); } finally { @@ -245,7 +250,7 @@ export async function detachNativeSessionsForRestart( const unsupportedRunIds: string[] = []; for (const runId of new Set(runIds)) { nativeRunsDetachingForRestart.add(runId); - const active = activeNativeSessions.get(runId) ?? await waitForNativeStartupForRestart(runId); + const active = activeNativeSessions.get(runId) ?? await waitForNativeSessionStartup(runId); if (!active) { inactiveRunIds.push(runId); continue; @@ -6289,6 +6294,7 @@ export async function cancelNativeSession( auditId: string | null; } > { + const runStop = (options?.scope ?? "run") === "run"; let decision: NativeStatusDecision | null = null; let decisionContext: { companyId: string; @@ -6554,7 +6560,8 @@ export async function cancelNativeSession( decisionId = intent.decisionId; recoveringCancellationIntent = intent.existing; priorCoordinatorDecisionIdAtIntent = intent.priorCoordinatorDecisionId; - if (intent.acknowledged) { + if (intent.acknowledged && (!runStop || intent.dispatched || + (!nativeSessionStartups.has(runId) && !activeNativeSessions.has(runId)))) { return { dispatched: intent.dispatched, decision, @@ -6563,150 +6570,208 @@ export async function cancelNativeSession( }; } } - const active = activeNativeSessions.get(runId); - let dispatched = false; - if (active) { - dispatched = true; - if (!active.cancelRequested) { - active.cancelRequested = true; - try { - if (active.session.cancel) { - const cancellationAbort = new AbortController(); - const cleanup = active.session.cancel({ - reason, - signal: cancellationAbort.signal, - }).cleanup; - let cleanupTimer: ReturnType | undefined; - const settled = await Promise.race([ - cleanup.then( - () => true, - () => true, - ), - new Promise((resolve) => { - cleanupTimer = setTimeout( - () => resolve(false), - NATIVE_SESSION_CANCELLATION_CLEANUP_GRACE_MS, - ); - }), - ]); - if (cleanupTimer) clearTimeout(cleanupTimer); - if (!settled) { - cancellationAbort.abort( - new Error("native session cancellation cleanup timed out"), - ); - void cleanup.catch(() => undefined); - } - } else if (active.session.interrupt) - await active.session.interrupt({ reason }); - } catch (error) { - active.cancelRequested = false; - throw error; - } - } + // Startup already owns a provider lifetime even before publishing its handle. + // Do not acknowledge a no-op Stop and let that owner submit a turn afterwards. + const startup = runStop && !activeNativeSessions.has(runId) ? nativeSessionStartups.get(runId) : undefined; + let settleStartupCancellation: (() => void) | undefined; + if (startup) { + startup.stopRequested = true; + startup.cancellationSettled = new Promise(resolve => { settleStartupCancellation = resolve; }); } - if (!options) return dispatched; - - if (decision && decisionContext) { - const cancellationDecision = decision; - const cancellationContext = decisionContext; - if (!cancellationIntentId || !auditId) - throw new Error("native_cancellation_intent_audit_missing"); - if ( - recoveringCancellationIntent && - cancellationContext.coordinatorDecisionId !== - priorCoordinatorDecisionIdAtIntent - ) { - decisionId ??= cancellationContext.coordinatorDecisionId; - } - if ( - cancellationContext.assessmentId && - cancellationDecision.reasonCode !== null && - !decisionId - ) { - const committed = await commitNativeStatusDecision({ - db: options.db, - companyId: cancellationContext.companyId, - issueId: cancellationContext.issueId, - runId, - assessmentId: cancellationContext.assessmentId, - priorStatus: cancellationContext.priorStatus, - priorStatusVersion: cancellationContext.priorStatusVersion, - priorDecisionId: cancellationContext.priorDecisionId, - decision: cancellationDecision, - }); - decisionId = committed.decision.id; - } - const acknowledgement = await options.db.transaction(async (tx) => { - const lockedRun = await tx - .select({ - agentId: heartbeatRuns.agentId, - companyId: heartbeatRuns.companyId, - nativeIssueId: heartbeatRuns.nativeIssueId, - resultJson: heartbeatRuns.resultJson, - runtimeMode: heartbeatRuns.runtimeMode, - }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .for("update") - .limit(1) - .then((rows) => rows[0] ?? null); - if ( - !lockedRun || - lockedRun.runtimeMode !== "native" || - lockedRun.companyId !== cancellationContext.companyId || - lockedRun.agentId !== cancellationContext.agentId || - lockedRun.nativeIssueId !== cancellationContext.issueId - ) { - throw new Error("native_cancellation_binding_changed"); - } - const coordinator = await tx - .select({ runId: nativeRunFinalizations.runId }) - .from(nativeRunFinalizations) - .where( - and( - eq(nativeRunFinalizations.runId, runId), - eq(nativeRunFinalizations.companyId, cancellationContext.companyId), - eq(nativeRunFinalizations.issueId, cancellationContext.issueId), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - if (!coordinator) - throw new Error("native_cancellation_coordinator_missing"); - - const resultJson = record(lockedRun.resultJson); - const intent = record(resultJson.nativeCancellation); - const matchingIntent = - intent.schema === "paperclip.native-cancellation.v1" && - intent.intentId === cancellationIntentId && - intent.intentAuditId === auditId && - intent.companyId === cancellationContext.companyId && - intent.runId === runId && - intent.issueId === cancellationContext.issueId; - if (!matchingIntent) - throw new Error("native_cancellation_intent_conflict"); - if (intent.dispatchState === "acknowledged") { - return { - publication: null, - decisionId: - typeof intent.decisionId === "string" - ? intent.decisionId - : decisionId, - }; + try { + const active = activeNativeSessions.get(runId) ?? + (startup ? await waitForNativeSessionStartup(runId, () => new NativeCancellationPendingRecoveryError()) : null); + let dispatched = false; + if (active) { + dispatched = true; + if (!active.cancelRequested) { + active.cancelRequested = true; + try { + if (active.session.cancel) { + const cancellationAbort = new AbortController(); + const cleanup = active.session.cancel({ + reason, + signal: cancellationAbort.signal, + }).cleanup; + let cleanupTimer: ReturnType | undefined; + const settled = await Promise.race([ + cleanup.then( + () => true, + () => true, + ), + new Promise((resolve) => { + cleanupTimer = setTimeout( + () => resolve(false), + NATIVE_SESSION_CANCELLATION_CLEANUP_GRACE_MS, + ); + }), + ]); + if (cleanupTimer) clearTimeout(cleanupTimer); + if (!settled) { + cancellationAbort.abort( + new Error("native session cancellation cleanup timed out"), + ); + void cleanup.catch(() => undefined); + } + } else if (active.session.interrupt) + await active.session.interrupt({ reason }); + } catch (error) { + active.cancelRequested = false; + throw error; + } } + } + if (!options) return dispatched; + if (decision && decisionContext) { + const cancellationDecision = decision; + const cancellationContext = decisionContext; + if (!cancellationIntentId || !auditId) + throw new Error("native_cancellation_intent_audit_missing"); if ( - options.replacementAccepted && - cancellationDecision.effects.some( - (effect) => effect.kind === "accept_replacement_turn", - ) + recoveringCancellationIntent && + cancellationContext.coordinatorDecisionId !== + priorCoordinatorDecisionIdAtIntent ) { - await tx + decisionId ??= cancellationContext.coordinatorDecisionId; + } + if ( + cancellationContext.assessmentId && + cancellationDecision.reasonCode !== null && + !decisionId + ) { + const committed = await commitNativeStatusDecision({ + db: options.db, + companyId: cancellationContext.companyId, + issueId: cancellationContext.issueId, + runId, + assessmentId: cancellationContext.assessmentId, + priorStatus: cancellationContext.priorStatus, + priorStatusVersion: cancellationContext.priorStatusVersion, + priorDecisionId: cancellationContext.priorDecisionId, + decision: cancellationDecision, + }); + decisionId = committed.decision.id; + } + const acknowledgement = await options.db.transaction(async (tx) => { + const lockedRun = await tx + .select({ + agentId: heartbeatRuns.agentId, + companyId: heartbeatRuns.companyId, + nativeIssueId: heartbeatRuns.nativeIssueId, + resultJson: heartbeatRuns.resultJson, + runtimeMode: heartbeatRuns.runtimeMode, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null); + if ( + !lockedRun || + lockedRun.runtimeMode !== "native" || + lockedRun.companyId !== cancellationContext.companyId || + lockedRun.agentId !== cancellationContext.agentId || + lockedRun.nativeIssueId !== cancellationContext.issueId + ) { + throw new Error("native_cancellation_binding_changed"); + } + const coordinator = await tx + .select({ runId: nativeRunFinalizations.runId }) + .from(nativeRunFinalizations) + .where( + and( + eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.companyId, cancellationContext.companyId), + eq(nativeRunFinalizations.issueId, cancellationContext.issueId), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!coordinator) + throw new Error("native_cancellation_coordinator_missing"); + + const resultJson = record(lockedRun.resultJson); + const intent = record(resultJson.nativeCancellation); + const matchingIntent = + intent.schema === "paperclip.native-cancellation.v1" && + intent.intentId === cancellationIntentId && + intent.intentAuditId === auditId && + intent.companyId === cancellationContext.companyId && + intent.runId === runId && + intent.issueId === cancellationContext.issueId; + if (!matchingIntent) + throw new Error("native_cancellation_intent_conflict"); + if (intent.dispatchState === "acknowledged" && (intent.dispatched === true || !dispatched)) { + return { + publication: null, + decisionId: + typeof intent.decisionId === "string" + ? intent.decisionId + : decisionId, + }; + } + + if ( + options.replacementAccepted && + cancellationDecision.effects.some( + (effect) => effect.kind === "accept_replacement_turn", + ) + ) { + await tx + .update(heartbeatRuns) + .set({ + status: "running", + continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`, + nextAction: "Accept a replacement native turn on the existing run.", + updatedAt: new Date(), + }) + .where( + and( + eq(heartbeatRuns.id, runId), + eq(heartbeatRuns.companyId, cancellationContext.companyId), + eq(heartbeatRuns.agentId, cancellationContext.agentId), + eq(heartbeatRuns.nativeIssueId, cancellationContext.issueId), + ), + ); + } + const activity = await persistActivity(tx as unknown as Db, { + companyId: cancellationContext.companyId, + actorType: "system", + actorId: "native-session-cancellation", + action: "native.cancellation_dispatch_acknowledged", + entityType: "heartbeat_run", + entityId: runId, + agentId: cancellationContext.agentId, + runId, + issueId: cancellationContext.issueId, + details: { + intentId: cancellationIntentId, + intentAuditId: auditId, + scope: options.scope ?? "run", + reasonCode: cancellationDecision.reasonCode, + effects: cancellationDecision.effects.map((effect) => effect.kind), + dispatched, + decisionId, + }, + }); + const acknowledgementAuditId = activity.activity?.id ?? null; + if (!acknowledgementAuditId) + throw new Error("native_cancellation_ack_audit_missing"); + const cancellationWrite = await tx .update(heartbeatRuns) .set({ - status: "running", - continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`, - nextAction: "Accept a replacement native turn on the existing run.", + resultJson: { + ...resultJson, + nativeCancellation: { + ...intent, + dispatchState: "acknowledged", + dispatched, + decisionId, + acknowledgementAuditId, + acknowledgedAt: new Date().toISOString(), + }, + }, updatedAt: new Date(), }) .where( @@ -6716,66 +6781,21 @@ export async function cancelNativeSession( eq(heartbeatRuns.agentId, cancellationContext.agentId), eq(heartbeatRuns.nativeIssueId, cancellationContext.issueId), ), - ); - } - const activity = await persistActivity(tx as unknown as Db, { - companyId: cancellationContext.companyId, - actorType: "system", - actorId: "native-session-cancellation", - action: "native.cancellation_dispatch_acknowledged", - entityType: "heartbeat_run", - entityId: runId, - agentId: cancellationContext.agentId, - runId, - issueId: cancellationContext.issueId, - details: { - intentId: cancellationIntentId, - intentAuditId: auditId, - scope: options.scope ?? "run", - reasonCode: cancellationDecision.reasonCode, - effects: cancellationDecision.effects.map((effect) => effect.kind), - dispatched, - decisionId, - }, + ) + .returning({ id: heartbeatRuns.id }) + .then((rows) => rows[0] ?? null); + if (!cancellationWrite) + throw new Error("native_cancellation_binding_changed"); + return { publication: activity.publication, decisionId }; }); - const acknowledgementAuditId = activity.activity?.id ?? null; - if (!acknowledgementAuditId) - throw new Error("native_cancellation_ack_audit_missing"); - const cancellationWrite = await tx - .update(heartbeatRuns) - .set({ - resultJson: { - ...resultJson, - nativeCancellation: { - ...intent, - dispatchState: "acknowledged", - dispatched, - decisionId, - acknowledgementAuditId, - acknowledgedAt: new Date().toISOString(), - }, - }, - updatedAt: new Date(), - }) - .where( - and( - eq(heartbeatRuns.id, runId), - eq(heartbeatRuns.companyId, cancellationContext.companyId), - eq(heartbeatRuns.agentId, cancellationContext.agentId), - eq(heartbeatRuns.nativeIssueId, cancellationContext.issueId), - ), - ) - .returning({ id: heartbeatRuns.id }) - .then((rows) => rows[0] ?? null); - if (!cancellationWrite) - throw new Error("native_cancellation_binding_changed"); - return { publication: activity.publication, decisionId }; - }); - decisionId = acknowledgement.decisionId; - if (acknowledgement.publication) - publishActivity(acknowledgement.publication); + decisionId = acknowledgement.decisionId; + if (acknowledgement.publication) + publishActivity(acknowledgement.publication); + } + return { dispatched, decision, decisionId, auditId }; + } finally { + settleStartupCancellation?.(); } - return { dispatched, decision, decisionId, auditId }; } /** Authenticated cancellation scope projected through the shared arbiter. */ @@ -8164,7 +8184,18 @@ async function executePaperclipNativeSessionWithinScope( if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { if (session.detachControllerForRestart) await detachActiveNativeSessionForRestart(active); } - nativeSessionStartups.get(input.execution.binding.runId)?.resolve(active); + const startup = nativeSessionStartups.get(input.execution.binding.runId); + startup?.resolve(active); + if (startup?.stopRequested) { + // Publication wakes the Stop caller; wait for its durable ACK + // before unwinding. No provider turn may be submitted between + // session startup and the execution-owned cleanup below. + await startup.cancellationSettled; + // If startup exceeded the caller's deadline, its late handle + // must still be cancelled instead of escaping the Stop fence. + await cancelNativeSession(input.execution.binding.runId, "Stop requested during native startup"); + throw new Error("native_finalization_missing: session returned no semantic result"); + } } else { liveQuestions.close(); activeNativeSessions.delete(input.execution.binding.runId); diff --git a/tests/runner-e2e/FIXTURES.md b/tests/runner-e2e/FIXTURES.md index 990ec383670..f03bfda838e 100644 --- a/tests/runner-e2e/FIXTURES.md +++ b/tests/runner-e2e/FIXTURES.md @@ -158,9 +158,9 @@ manual or scheduled campaign, not a PR requirement. ## Persistent chat fixtures -`chat-cases.ts` defines the six-case `agent-chat` suite; `chat-flow.ts` drives the +`chat-cases.ts` defines the eight-case `agent-chat` suite; `chat-flow.ts` drives the production composer, plan revision/approval controls, questions, reset command, -and project cards. Keep its 24 local cells intentional. `expectedRunCount` +and project cards. Keep its 28 local cells intentional. `expectedRunCount` counts provider turns, including cancelled and handed-off task runs, but excludes synthetic `/new` runs. Assertions must inspect all company runs because ordinary issue lists exclude the source conversation. `assertChatHandoff` rejects missing @@ -172,3 +172,18 @@ workspaces, task documents, and ordering. They pass through the normal sanitizer Screenshots are allowlisted to the exact disposable agent chat. Cleanup cancels all active runs in the isolated company, including handed-off work; usage from failed and cancelled runs must not disappear from campaign totals. + +`chat-hardening.ts` adds the explicit-only `agent-chat-hardening` journeys. Use +the ordinary public APIs to seed source documents and blockers. Keep the answer +out of the user's status/review request. Grade the exact source values, latest +blocker, preserved task identities, worker-authored output, and real executions. +The status request asks for JSON so the grader can distinguish the current +blocker from a historical mention and compare active-run count separately from +task status. The request must not reveal those expected values. +Capture the source after seeding and compare every field in the public issue +update contract, plus labels, dependencies, and dedicated-endpoint settings. +Derived inbound references may change when the chat legitimately cites a task. +The lost-acknowledgement probe may interrupt only the fixture browser's own +comment request after the real server has committed it. Retain its request ID +and replay that same request through the public API after restarting the server. +Never fabricate tool results or repair task state after a failed assertion. diff --git a/tests/runner-e2e/README.md b/tests/runner-e2e/README.md index 05ea24825a5..29f5694a7bb 100644 --- a/tests/runner-e2e/README.md +++ b/tests/runner-e2e/README.md @@ -131,15 +131,15 @@ runner instance, PID, and process-start identity. Each turn is bounded to ten minutes, the cell to thirty minutes, and cleanup explicitly deletes the sandbox rather than waiting for Daytona's idle timeout. -`agent-chat` (**Persistent Agent Chat**) adds six workflows on `legacy-codex`, -`legacy-claude`, `runner-codex`, and `runner-acpx-claude`: **26 local cells**. +`agent-chat` (**Persistent Agent Chat**) has eight workflows on `legacy-codex`, +`legacy-claude`, `runner-codex`, and `runner-acpx-claude`: **28 local cells**. They cover continuity across server restart, fresh context after `/new`, Stop/reset/resume, draft/revise/approve/plan handoff, clarification with existing project reuse, and a new project with two repository URLs. Each cell opens the production chat surface and resolves the backing issue through the chat API. The source conversation must settle to `in_review` / `waiting`; handed-off execution tasks must finish with their initial Plan and output documents. -Reset runs are retained separately from the 72 expected provider turns in this +Reset runs are retained separately from the 76 expected run attempts in this suite. Cancelled turns and execution-task runs remain included in billing and cleanup. The production chat directive is injected normally; fixtures do not replace it with completion instructions. Daytona is excluded. @@ -151,6 +151,48 @@ and backlog tasks. The oracle verifies stable task IDs, preserved descriptions, assignment audit evidence, exactly one successful successor run and its output document, no backlog execution, and a usable source conversation after reload. +`agent-chat-hardening` is an explicit-only native Codex/Claude suite with +**18 cells: 12 local and six warm Daytona**. It adds startup Stop/reset, +hire/delegate/reuse, grounded blocker reporting with source-document review, +and committed-send retry. It also runs active Stop/reset and restart continuity. +Daytona selects active Stop/reset, restart continuity, and committed-send retry. +The 56 expected run attempts include cancelled attempts; synthetic resets are +recorded separately. The suite uses production permission defaults and prompts. +Only hiring and cross-task status/review enable the opt-in native API tools. +Hiring uses a personal managed AI connection and verifies the hired worker's +actual execution account. This is not an onboarding-default qualification. + +Stop during startup and Stop during an active response are separate boundaries. +The native active-response case requires a recorded provider turn start; generic +lifecycle/performance events are insufficient. The startup case must stop after +a process launch request but before a provider turn starts. Missing the boundary +fails the case instead of silently testing another phase. + +Restart continuity requires the agent to recall a phrase after the server +restarts. The final prompt does not reveal that phrase. A generic successful +reply after restart cannot pass this check. + +The blocker query requests a JSON status snapshot. It must name the current +recorded blocker and report zero active runs independently of the task's blocked +status. Mentioning the right blocker only as resolved history cannot pass. + +The committed-send case drops the browser's acknowledgement after the server +saves its comment. It waits for the agent to save one backlog task, restarts +Paperclip, and replays the exact public request with the original client request ID. It +requires the original comment, task, plan, and single consuming run. This proves +HTTP request idempotency across restart, not replay safety for an ambiguous +provider tool response. Existing native tool-receipt tests cover that boundary. + +```sh +pnpm test:e2e:runner -- --list --suite agent-chat-hardening +pnpm test:e2e:runner -- --id agent-chat-hardening.runner-codex.local.stop-startup-new-resume +``` + +Each hardening oracle has positive and plausible-negative calibration tests. +The review grader parses the worker's saved JSON and compares both source values +and the consistency verdict. Hiring requires one identity, correct reporting +line, managed credentials, and real task execution; chat claims cannot pass it. + ```bash # Run these after deterministic checks, with the required provider keys set. pnpm test:e2e:runner -- --id agent-chat.legacy-codex.local.continuity-restart @@ -176,8 +218,9 @@ throwaway instance; never point the authenticated suite at the running demo. Missing provider credentials fail paid preflight and are not passing coverage. The default `--all` selection is 171 cells (148 local and 23 Daytona) and 371 -expected paid agent turns. The explicit-only everyday suite adds 35 catalog cells -and is excluded from `--all`. Follow-up steps remain ordered within their cell; all other +expected paid agent turns. The explicit-only everyday suite adds 38 catalog cells +and chat hardening adds 18. Both are excluded from `--all`. The full catalog has 227 cells. +Follow-up steps remain ordered within their cell; all other cells are independent. Narrow selectors are strongly recommended while developing fixtures. diff --git a/tests/runner-e2e/catalog.test.ts b/tests/runner-e2e/catalog.test.ts index c6772a0680f..027ab20201d 100644 --- a/tests/runner-e2e/catalog.test.ts +++ b/tests/runner-e2e/catalog.test.ts @@ -71,10 +71,10 @@ describe("runner E2E catalog", () => { expect(localIntegrityTasks).toHaveLength(2); expect(openRouterBreadthTasks).toHaveLength(3); expect(runnerSuites.map((suite) => suite.expectedMatrixSize)).toEqual([ - 23, 38, 52, 28, 42, 14, 10, 2, + 23, 38, 52, 28, 18, 42, 14, 10, 2, ]); - expect(validateRunnerCatalog()).toHaveLength(209); - expect(new Set(runnerMatrix.map((entry) => entry.id)).size).toBe(209); + expect(validateRunnerCatalog()).toHaveLength(227); + expect(new Set(runnerMatrix.map((entry) => entry.id)).size).toBe(227); expect( runnerMatrix.filter((entry) => entry.suite.id === "core-compatibility"), ).toHaveLength(42); diff --git a/tests/runner-e2e/catalog.ts b/tests/runner-e2e/catalog.ts index 0686733664b..73b6cd99272 100644 --- a/tests/runner-e2e/catalog.ts +++ b/tests/runner-e2e/catalog.ts @@ -2,7 +2,7 @@ import { continuationTasks } from "./continuation-cases.js"; import { everydayTasks, productionStoryProfile } from "./everyday-cases.js"; import { firstTaskTasks } from "./first-task-cases.js"; -import { chatTasks } from "./chat-cases.js"; +import { chatTasks, chatHardeningTasks } from "./chat-cases.js"; import { createHash } from "node:crypto"; import { createAgentSchema } from "../../packages/shared/src/validators/agent.js"; import { createEnvironmentSchema } from "../../packages/shared/src/validators/environment.js"; @@ -948,7 +948,18 @@ export const runnerSuites: readonly RunnerSuiteFixture[] = [ profiles: runnerProfiles.filter(profile => ["legacy-codex", "legacy-claude", "runner-codex", "runner-acpx-claude"].includes(profile.id)).map(defaultPermissionProfile), environments: [localEnvironment], tasks: chatTasks, expectedMatrixSize: 28, excludedExecutionIds: ["legacy-codex", "legacy-claude"].flatMap(profile => ["reassign-task", "create-backlog"].map(task => `agent-chat.${profile}.local.${task}`)), - definitionMetadata: { version: 4, resetRunsCountedSeparately: true, permissions: "production-defaults" }, + definitionMetadata: { version: 6, resetRunsCountedSeparately: true, permissions: "production-defaults", stopBoundary: "provider-turn-started", restartMemory: "required-after-restart" }, + }, + { + id: "agent-chat-hardening", label: "Agent Chat Recovery and Coordination", manualOnly: true, + description: "Native chat startup cancellation, committed sends, hiring, grounded status, and remote continuity.", + groups: ["chat", "native"], + profiles: runnerProfiles.filter(profile => ["runner-codex", "runner-acpx-claude"].includes(profile.id)) + .map(profile => productionStoryProfile(defaultPermissionProfile(profile))), + environments: [localEnvironment, daytonaWarmEnvironment], tasks: chatHardeningTasks, expectedMatrixSize: 18, + excludedExecutionIds: ["runner-codex", "runner-acpx-claude"].flatMap(profile => + ["stop-startup-new-resume", "hire-delegate-reuse", "blocked-status-review"].map(task => `agent-chat-hardening.${profile}.daytona.${task}`)), + definitionMetadata: { version: 4, permissions: "production-defaults", instructions: "production", grading: "durable-state-and-source-evidence", scheduling: "explicit-only", restartMemory: "required-after-restart", statusEvidence: "structured-current-blocker-and-active-run-count", readOnlyState: "public-mutation-contract-and-relations" }, }, ...(process.env.PAPERCLIP_RUNNER_E2E_CONNECTION_REVIEWS === "1" ? [connectionReviewSuite] : []), { diff --git a/tests/runner-e2e/chat-cases.ts b/tests/runner-e2e/chat-cases.ts index a8f2cd31d70..c0e59620e08 100644 --- a/tests/runner-e2e/chat-cases.ts +++ b/tests/runner-e2e/chat-cases.ts @@ -20,7 +20,15 @@ export const CHAT_CASES = [ ["multi-repository", "Create a project with multiple repository URLs", 2], ] as const; export type ChatCase = (typeof CHAT_CASES)[number][0]; -export const chatTasks: readonly RunnerTaskFixture[] = CHAT_CASES.map( +const HARDENING_CASES = [ + ["stop-startup-new-resume", "Stop during startup, reset, and resume", 3], + ["hire-delegate-reuse", "Hire through chat, delegate, and reuse the same teammate", 5], + ["blocked-status-review", "Read the actual blocker and hand source material to a reviewer", 4], + ["committed-send-retry", "Recover a lost send acknowledgement without repeating committed work", 2], + ...CHAT_CASES.filter(([id]) => ["stop-new-resume", "continuity-restart"].includes(id)), +] as const; +function buildChatTasks(definitions: readonly (readonly [string, string, number])[]): readonly RunnerTaskFixture[] { + return definitions.map( ([id, label, expectedRunCount]) => ({ id, label, @@ -35,4 +43,15 @@ export const chatTasks: readonly RunnerTaskFixture[] = CHAT_CASES.map( buildVisibleMarker: (nonce) => chatMarker("CHAT", nonce), buildMatchers: () => [{ kind: "issue_status", expected: "in_review" }], }), -); + ); +} +export const chatTasks = buildChatTasks(CHAT_CASES); +export const chatHardeningTasks = buildChatTasks(HARDENING_CASES); + +export function chatNeedsApiTools(suiteId: string, caseId: string): boolean { + return suiteId === "agent-chat-hardening" && ["hire-delegate-reuse", "blocked-status-review"].includes(caseId); +} +export function isManagedHiringCase(suiteId: string, caseId: string): boolean { + return (suiteId === "everyday-workflows" && caseId === "hire-reuse") || + (suiteId === "agent-chat-hardening" && caseId === "hire-delegate-reuse"); +} diff --git a/tests/runner-e2e/chat-flow.ts b/tests/runner-e2e/chat-flow.ts index 24d5aa9233c..99fd8dfa606 100644 --- a/tests/runner-e2e/chat-flow.ts +++ b/tests/runner-e2e/chat-flow.ts @@ -8,6 +8,7 @@ import type { LiveFixtureValues } from "./live-fixtures.js"; import type { MatrixExecution } from "./types.js"; import { isBlockedUnstartedWake } from "./non-execution-wake.js"; import { chatMarker } from "./chat-cases.js"; +import { assertChatRememberedAfterRestart, assertChatStartupStopped, isChatStopReady, runChatHardeningFlow } from "./chat-hardening.js"; // Public API observations only: this driver never fabricates provider results or writes DB state. export interface ChatIssue { @@ -285,7 +286,7 @@ export function assertChatReassignment(input: { expect(input.outputBody).toContain(input.marker); } -export async function runChatFlow(input: { +export interface ChatFlowInput { page: Page; api: RunnerApi; fixtures: LiveFixtureValues; @@ -296,13 +297,15 @@ export async function runChatFlow(input: { observe: (issue: ChatIssue, runs: ChatRun[]) => void; capture: (id: string, label: string, file: string) => Promise; evidence: (name: string, data: unknown) => Promise; -}) { +} +export async function runChatFlow(input: ChatFlowInput) { const { page, api, fixtures: f, execution, nonce } = input; const chatPath = `/api/companies/${f.company.id}/chats/${f.agent.id}`; const route = `/${f.company.issuePrefix}/chats/${f.agent.id}`; const marker = execution.task.buildVisibleMarker(nonce); const draftMarker = chatMarker("DRAFT", nonce); const caseId = execution.task.id; + const stopCase = ["stop-new-resume", "stop-startup-new-resume"].includes(caseId); let issue: ChatIssue; let runs: ChatRun[] = []; const settings = await api.get>( @@ -346,7 +349,7 @@ export async function runChatFlow(input: { activeRuns: runs .filter((run) => ["queued", "running"].includes(run.status)) .map((run) => run.id), - failure: chatRunFailure(runs, caseId === "stop-new-resume"), + failure: chatRunFailure(runs, stopCase), }; }, reject: (state) => state.failure ?? inconsistentIdle(state), @@ -378,7 +381,7 @@ export async function runChatFlow(input: { expect(await allRuns()).toHaveLength(0); if ( - ["continuity-restart", "new-session", "stop-new-resume"].includes(caseId) + ["continuity-restart", "new-session", "stop-new-resume", "stop-startup-new-resume"].includes(caseId) ) { const secret = chatMarker("OLDCONTEXT", nonce); await turn( @@ -401,17 +404,24 @@ export async function runChatFlow(input: { await input.restart(); // Re-enter the canonical route after the server replaces its browser // transport; reloading the stale document can target a detached page. - await page.goto(route, { waitUntil: "domcontentloaded", timeout: 60_000 }); - await expect(page.getByTestId("task-chat-composer-input")).toBeVisible(); + // The restarted dev server can leave DOMContentLoaded pending after the + // app is interactive. Require the actual chat composer after navigation. + await page.goto(route, { waitUntil: "commit", timeout: 60_000 }); + await expect(page.getByTestId("task-chat-composer-input")).toBeVisible({ timeout: 60_000 }); await idle(2); expect(runs).toHaveLength(count); await turn( - `We are done discussing it. Reply with ${marker} only; no further work.`, + `What phrase did I ask you to remember earlier? Reply with that remembered phrase followed by ${marker}; no further work.`, 3, ); + assertChatRememberedAfterRestart( + (await comments()).filter((comment) => comment.authorAgentId).at(-1)?.body ?? "", + secret, + marker, + ); } else { let cancelledId: string | undefined; - if (caseId === "stop-new-resume") { + if (stopCase) { await sendChatMessage( page, "Explain the history of gardening at length here, in 100 numbered paragraphs. This is discussion only; do not create work.", @@ -427,9 +437,14 @@ export async function runChatFlow(input: { const events = await api.get>>( `/api/heartbeat-runs/${active.id}/events?limit=1000`, ); - const log = await readRunningChatLog(api, active.id); - if (!(events.length || log?.length)) return false; + if (execution.profile.generation === "native") { + if (!isChatStopReady(events, caseId === "stop-startup-new-resume" ? "startup" : "active")) return false; + } else if (!(events.length || (await readRunningChatLog(api, active.id))?.length)) return false; cancelledId = active.id; + await input.evidence("chat-stop-boundary.json", { + phase: caseId === "stop-startup-new-resume" ? "startup" : "active", + runId: active.id, events, + }); return true; }, { timeout: 120_000 }, @@ -441,6 +456,7 @@ export async function runChatFlow(input: { async () => (await api.get(`/api/heartbeat-runs/${cancelledId}`)) .status, + { timeout: 30_000 }, ) .toBe("cancelled"); } @@ -495,6 +511,12 @@ export async function runChatFlow(input: { ).toEqual( oldComments.filter((c) => c.createdByRunId === cancelledId), ); + if (caseId === "stop-startup-new-resume" && cancelledId) { + const stopped = await api.get(`/api/heartbeat-runs/${cancelledId}`); + const events = await api.get>(`/api/heartbeat-runs/${cancelledId}/events?limit=1000`); + await input.evidence("chat-startup-stop-result.json", { stopped, events }); + assertChatStartupStopped(stopped, events); + } await page.reload({ waitUntil: "domcontentloaded", timeout: 60_000 }); await expect( page.getByText("New session", { exact: true }), @@ -502,6 +524,8 @@ export async function runChatFlow(input: { } expect(issue!.id).toBe(initialId); await noTasks(); + } else if (["hire-delegate-reuse", "blocked-status-review", "committed-send-retry"].includes(caseId)) { + await runChatHardeningFlow({ input, marker, issue: () => issue!, turn, idle, tasks, allRuns, comments }); } else if (caseId === "create-backlog") { const project = await api.post<{ id: string }>(`/api/companies/${f.company.id}/projects`, { name: `Later planning ${nonce}`, description: "Repository-free plans to save for later.", @@ -894,7 +918,7 @@ export async function runChatFlow(input: { for (const run of runs.filter((run) => !isResetRun(run))) { expect(run.runtimeMode).toBe(execution.profile.expectedRuntimeMode); expect(run.status).toBe( - caseId === "stop-new-resume" && run.status === "cancelled" + stopCase && run.status === "cancelled" ? "cancelled" : "succeeded", ); diff --git a/tests/runner-e2e/chat-hardening.test.ts b/tests/runner-e2e/chat-hardening.test.ts new file mode 100644 index 00000000000..5ebcadc89a3 --- /dev/null +++ b/tests/runner-e2e/chat-hardening.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, it } from "vitest"; +import { assertChatHire, assertChatRememberedAfterRestart, assertChatSourceReview, assertChatStartupStopped, assertCommittedSendRetry, assertGroundedChatStatus, isChatStopReady } from "./chat-hardening.js"; +import { runnerMatrix } from "./catalog.js"; +import { buildRunnerE2EProcessEnvironment } from "./harness-env.js"; + +const binding = { provider: "anthropic", method: "api_key", mode: "responsible_user" }; +const hire = { id: "hire", name: "Morgan", reportsTo: "lead", adapterType: "paperclip_runner", adapterConfig: { model: "model" }, runtimeConfig: { aiConnection: binding } }; +const task = { id: "task", companyId: "company", title: "Checklist", status: "done", assigneeAgentId: "hire", parentId: null, projectId: "project" }; +const run = { id: "run", companyId: "company", agentId: "hire", status: "succeeded", runtimeMode: "native", contextSnapshot: { issueId: "task", aiConnection: { connectionId: "account" } } }; +const hiring = { agents: [{ id: "lead", name: "Lead", adapterConfig: { model: "model" } }, hire], leadId: "lead", hireName: "Morgan", hiredId: "hire", connectionId: "account", binding, taskIds: ["task"], tasks: [task], runs: [run] }; + +describe("agent chat hardening oracles", () => { + it("requires remembered context after restart, not just a new reply", () => { + expect(() => assertChatRememberedAfterRestart("OLDCONTEXT123 CHAT123", "OLDCONTEXT123", "CHAT123")).not.toThrow(); + expect(() => assertChatRememberedAfterRestart("CHAT123", "OLDCONTEXT123", "CHAT123")).toThrow(); + expect(() => assertChatRememberedAfterRestart("OLDCONTEXT999 CHAT123", "OLDCONTEXT123", "CHAT123")).toThrow(); + }); + it("grades actual execution by the original hired identity and managed account", () => { + expect(() => assertChatHire(hiring)).not.toThrow(); + for (const wrong of [ + { ...hiring, agents: [...hiring.agents, { ...hire, id: "duplicate" }] }, + { ...hiring, hiredId: "replaced" }, + { ...hiring, tasks: [{ ...task, parentId: "chat" }] }, + { ...hiring, tasks: [{ ...task, assigneeAgentId: "lead" }] }, + { ...hiring, runs: [] }, + { ...hiring, runs: [run, { ...run, id: "duplicate-run" }] }, + { ...hiring, runs: [{ ...run, agentId: "lead" }] }, + { ...hiring, connectionId: "wrong-account" }, + ]) expect(() => assertChatHire(wrong)).toThrow(); + }); + + it("requires the current recorded blocker without silently starting or changing work", () => { + const before = { ...task, status: "blocked" }; + const status = { issueIdentifier: "RUN-2", status: "blocked", currentBlockerLabel: "VENUE123", activeRunCount: 0 }; + const valid = { reply: JSON.stringify(status), expectedIssueIdentifier: "RUN-2", blocker: "VENUE123", + before, after: before, taskIdsBefore: ["task"], taskIdsAfter: ["task"], taskRuns: [] }; + expect(() => assertGroundedChatStatus(valid)).not.toThrow(); + const cited = { ...before, relatedWork: { inbound: [{ sourceIssueId: "chat" }] } }; + expect(() => assertGroundedChatStatus({ ...valid, after: cited })).not.toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, reply: JSON.stringify({ ...status, explanation: "BUDGET123 was resolved." }) })).not.toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, reply: JSON.stringify({ ...status, currentBlockerLabel: "BUDGET123", explanation: "VENUE123 was resolved." }) })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, reply: JSON.stringify({ ...status, activeRunCount: 1 }) })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, reply: JSON.stringify({ ...status, status: "in_progress", explanation: "Previously blocked." }) })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, reply: JSON.stringify({ ...status, activeRunCount: undefined }) })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, reply: "RUN-2 is blocked on BUDGET123." })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, after: task })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, after: { ...before, title: "Changed" } })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, after: { ...before, projectId: "different" } })).toThrow(); + for (const mutation of [ + { description: "Changed the brief" }, { priority: "high" }, + { labelIds: ["new-label"] }, { labels: [{ id: "label", name: "Changed" }] }, + { blockedBy: [{ id: "new-blocker" }] }, { blocks: [{ id: "new-dependent" }] }, + { unblockDescriptor: { owner: "board", action: "Different action" } }, + { executionWorkspaceSettings: { mode: "isolated" } }, + { reviewPolicy: { kind: "none" } }, { watchdog: { agentId: "different" } }, + ]) expect(() => assertGroundedChatStatus({ ...valid, after: { ...before, ...mutation } })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, taskRuns: [run] })).toThrow(); + expect(() => assertGroundedChatStatus({ ...valid, taskIdsAfter: ["task", "replacement"] })).toThrow(); + }); + + it("independently compares review values and verdict, rejecting copied sources and missing evidence", () => { + const expected = { planLaunchDay: "Tuesday", briefLaunchDay: "Wednesday" }; + expect(() => assertChatSourceReview(JSON.stringify({ ...expected, consistent: false }), expected)).not.toThrow(); + expect(() => assertChatSourceReview(JSON.stringify({ ...expected, consistent: true }), expected)).toThrow(); + expect(() => assertChatSourceReview("The plan says Tuesday and the brief says Wednesday.", expected)).toThrow(); + expect(() => assertChatSourceReview("{}", expected)).toThrow(); + expect(() => assertChatSourceReview(JSON.stringify({ ...expected, planLaunchDay: "Friday", consistent: false }), expected)).toThrow(); + expect(() => assertChatSourceReview('{"planLaunchDay":"Tuesday","briefLaunchDay":"Tuesday","consistent":true}', { planLaunchDay: "Tuesday", briefLaunchDay: "Tuesday" })).not.toThrow(); + }); + + it("does not confuse lifecycle logs, provider startup, and an active turn", () => { + expect(isChatStopReady([{ eventType: "lifecycle" }, { eventType: "run.performance.span" }], "active")).toBe(false); + const startup = [{ eventType: "native.process_start_requested" }]; + expect(isChatStopReady(startup, "startup")).toBe(true); + expect(isChatStopReady(startup, "active")).toBe(false); + const active = [...startup, { eventType: "turn.started" }]; + expect(isChatStopReady(active, "startup")).toBe(false); + expect(isChatStopReady(active, "active")).toBe(true); + }); + + it("requires one committed comment, one task, and one consuming run after request replay", () => { + const valid = { commentId: "comment", clientRequestId: "request", comments: [{ id: "comment", body: "create", clientRequestId: "request" }], taskId: "task", + tasks: [{ ...task, status: "backlog" }], chatId: "chat", runs: [{ ...run, contextSnapshot: { issueId: "chat", wakeCommentId: "comment" } }] }; + expect(() => assertCommittedSendRetry(valid)).not.toThrow(); + expect(() => assertCommittedSendRetry({ ...valid, comments: [...valid.comments, { ...valid.comments[0]!, id: "duplicate" }] })).toThrow(); + expect(() => assertCommittedSendRetry({ ...valid, tasks: [...valid.tasks, { ...task, id: "duplicate" }] })).toThrow(); + expect(() => assertCommittedSendRetry({ ...valid, runs: [...valid.runs, run] })).toThrow(); + expect(() => assertCommittedSendRetry({ ...valid, runs: [...valid.runs, { ...valid.runs[0]!, id: "duplicate" }] })).toThrow(); + expect(() => assertCommittedSendRetry({ ...valid, runs: [] })).toThrow(); + }); + + it("requires dispatched Stop with no late turn and reports a missed startup checkpoint separately", () => { + const cancellation = { scope: "run", dispatchState: "acknowledged", dispatched: true, recordedAt: "2026-09-21T12:00:00Z" }; + const stopped = { ...run, status: "cancelled", resultJson: { nativeCancellation: cancellation } }; + expect(() => assertChatStartupStopped(stopped, [])).not.toThrow(); + expect(() => assertChatStartupStopped({ ...stopped, resultJson: { nativeCancellation: { ...cancellation, dispatched: false } } }, [])).toThrow(); + expect(() => assertChatStartupStopped(stopped, [{ eventType: "turn.started", createdAt: "2026-09-21T12:00:01Z" }])).toThrow(/must never submit/); + expect(() => assertChatStartupStopped(stopped, [{ eventType: "turn.started", createdAt: "2026-09-21T11:59:59Z" }])).toThrow(/Harness missed startup Stop boundary/); + }); + + it("keeps the paid hardening matrix explicit and limits API-tool opt-in to coordination", () => { + const cells = runnerMatrix.filter(cell => cell.suite.id === "agent-chat-hardening"); + expect(cells).toHaveLength(18); + expect(cells.every(cell => cell.suite.manualOnly && cell.profile.generation === "native")).toBe(true); + expect(cells.filter(cell => cell.environment.id === "daytona")).toHaveLength(6); + for (const cell of cells) { + expect(buildRunnerE2EProcessEnvironment({}, [cell]).PAPERCLIP_RUNNER_API_TOOLS_ENABLED).toBe( + ["hire-delegate-reuse", "blocked-status-review"].includes(cell.task.id) ? "true" : undefined); + const config = cell.profile.buildAgent({ executionId: "fixture", workspacePath: "/workspace", environmentId: "local", + environmentFixtureId: "local", secretRefs: { [cell.profile.credential]: { type: "secret_ref", secretId: "secret", version: "latest" } } }); + expect(config.role).toBe("ceo"); + expect(JSON.stringify(config.instructionsBundle)).not.toMatch(/mark the task done|paperclip_finish|POST \/api|PUT \/api/); + expect(config.adapterConfig).not.toHaveProperty("codexPermissionMode"); + expect(config.adapterConfig).not.toHaveProperty("acpxPermissionMode"); + } + }); +}); diff --git a/tests/runner-e2e/chat-hardening.ts b/tests/runner-e2e/chat-hardening.ts new file mode 100644 index 00000000000..3c0d547c29c --- /dev/null +++ b/tests/runner-e2e/chat-hardening.ts @@ -0,0 +1,244 @@ +import { expect } from "@playwright/test"; +import { updateIssueSchema } from "../../packages/shared/src/validators/issue.js"; +import type { RunnerApi } from "./api.js"; +import { readChatOutputDocument, sendChatMessage, type ChatFlowInput, type ChatIssue, type ChatRun } from "./chat-flow.js"; +import { dropChatSendAcknowledgement } from "./lost-send.js"; + +type Agent = { + id: string; name: string; reportsTo?: string; adapterType?: string; + adapterConfig?: Record; runtimeConfig?: Record; +}; +type Comment = { id: string; body: string; authorAgentId?: string; clientRequestId?: string }; +type Document = { id: string; body: string; latestRevisionId: string; createdByAgentId?: string }; + +function mutableIssueSnapshot(issue: ChatIssue) { + // Follow the public mutation contract as it grows. Include relationships and + // fields governed by dedicated endpoints; omit derived read projections such + // as inbound references, which a legitimate status reply can add to the chat. + const keys = new Set([...Object.keys(updateIssueSchema.shape), + "id", "companyId", "responsibleUserId", "labels", "blockedBy", "blocks", "watchdog", "sourceTrust"]); + const record = issue as unknown as Record; + return Object.fromEntries([...keys].map(key => [key, record[key]])); +} + +export function assertChatHire(input: { + agents: Agent[]; leadId: string; hireName: string; hiredId?: string; + connectionId: string; binding: unknown; taskIds: string[]; tasks: ChatIssue[]; runs: ChatRun[]; +}) { + const hires = input.agents.filter(agent => agent.name === input.hireName); + expect(hires).toHaveLength(1); + const hired = hires[0]!; + const lead = input.agents.find(agent => agent.id === input.leadId)!; + expect(input.connectionId).toBeTruthy(); + expect(hired).toMatchObject({ reportsTo: input.leadId, adapterType: "paperclip_runner" }); + if (input.hiredId) expect(hired.id).toBe(input.hiredId); + expect(hired.adapterConfig?.model).toBe(lead.adapterConfig?.model); + expect(hired.runtimeConfig?.aiConnection).toEqual(input.binding); + expect(input.tasks.map(task => task.id).sort()).toEqual([...input.taskIds].sort()); + for (const id of input.taskIds) { + const task = input.tasks.find(task => task.id === id)!; + expect(task).toMatchObject({ status: "done", assigneeAgentId: hired.id, parentId: null }); + expect(task.projectId).toBeTruthy(); + const runs = input.runs.filter(run => run.contextSnapshot?.issueId === id); + expect(runs).toHaveLength(1); + expect(runs[0]).toMatchObject({ agentId: hired.id, status: "succeeded", runtimeMode: "native" }); + expect(runs[0]!.contextSnapshot?.aiConnection).toMatchObject({ connectionId: input.connectionId }); + } + return hired; +} + +export function assertGroundedChatStatus(input: { + reply: string; expectedIssueIdentifier: string; blocker: string; + before: ChatIssue; after: ChatIssue; taskIdsBefore: string[]; taskIdsAfter: string[]; taskRuns: ChatRun[]; +}) { + // Mentioning a correct label is insufficient: the answer could falsely call + // it resolved, or confuse a blocked task with an actively running provider. + const json = input.reply.trim().replace(/^```(?:json)?\s*/, "").replace(/\s*```$/, ""); + expect(JSON.parse(json)).toMatchObject({ + issueIdentifier: input.expectedIssueIdentifier, status: "blocked", + currentBlockerLabel: input.blocker, activeRunCount: 0, + }); + expect(mutableIssueSnapshot(input.after), "Status reporting must preserve mutable issue state") + .toStrictEqual(mutableIssueSnapshot(input.before)); + expect([...input.taskIdsAfter].sort()).toEqual([...input.taskIdsBefore].sort()); + expect(input.taskRuns).toHaveLength(0); +} + +export function assertChatSourceReview(body: string, expected: { planLaunchDay: string; briefLaunchDay: string }) { + // The user requests JSON, so an independent parser grades the actual saved + // deliverable. A claim in the chat or a copied source document cannot pass. + const json = body.trim().replace(/^```(?:json)?\s*/, "").replace(/\s*```$/, ""); + expect(JSON.parse(json)).toMatchObject({ ...expected, consistent: expected.planLaunchDay === expected.briefLaunchDay }); +} + +export function assertCommittedSendRetry(input: { + commentId: string; clientRequestId: string; comments: Comment[]; taskId: string; + tasks: ChatIssue[]; runs: ChatRun[]; chatId: string; +}) { + expect(input.comments.filter(comment => comment.clientRequestId === input.clientRequestId).map(comment => comment.id)).toEqual([input.commentId]); + expect(input.tasks.map(task => task.id)).toEqual([input.taskId]); + expect(input.tasks[0]).toMatchObject({ status: "backlog", parentId: null }); + expect(input.runs.filter(run => run.contextSnapshot?.issueId === input.taskId)).toHaveLength(0); + const original = input.runs.filter(run => run.contextSnapshot?.wakeCommentId === input.commentId || + (Array.isArray(run.contextSnapshot?.wakeCommentIds) && run.contextSnapshot.wakeCommentIds.includes(input.commentId))); + expect(original).toHaveLength(1); + expect(original[0]).toMatchObject({ status: "succeeded", runtimeMode: "native" }); + expect(original[0]!.contextSnapshot?.issueId).toBe(input.chatId); +} + +export function isChatStopReady(events: Array<{ eventType?: unknown }>, phase: "startup" | "active"): boolean { + const active = events.some(event => ["turn.started", "turn.accepted"].includes(String(event.eventType))); + return phase === "active" ? active : !active && events.some(event => event.eventType === "native.process_start_requested"); +} + +export function assertChatRememberedAfterRestart(reply: string, remembered: string, marker: string) { + expect(reply).toContain(remembered); + expect(reply).toContain(marker); +} + +export function assertChatStartupStopped(run: ChatRun, events: Array<{ eventType?: unknown; createdAt?: string }>) { + const cancellation = run.resultJson?.nativeCancellation as Record | undefined; + const started = events.filter(event => ["turn.started", "turn.accepted"].includes(String(event.eventType))); + if (started.some(event => Date.parse(event.createdAt ?? "") < Date.parse(String(cancellation?.recordedAt)))) { + throw new Error("Harness missed startup Stop boundary: provider turn started before the Stop intent"); + } + expect(cancellation).toMatchObject({ scope: "run", dispatchState: "acknowledged", dispatched: true }); + expect(run.status).toBe("cancelled"); + expect(started, "A stopped startup must never submit a provider turn").toHaveLength(0); +} + +export async function runChatHardeningFlow(context: { + input: ChatFlowInput; marker: string; issue(): ChatIssue; + turn(message: string, count: number): Promise; idle(count: number): Promise; + tasks(): Promise; allRuns(): Promise; comments(): Promise; +}) { + const { input, marker, turn, idle, tasks, allRuns, comments } = context; + const { api, page, fixtures: f, execution, nonce } = input; + const companyPath = `/api/companies/${f.company.id}`; + const project = await api.post<{ id: string; name: string }>(`${companyPath}/projects`, { + name: `Chat coordination ${nonce}`, description: "Repository-free launch planning and review.", + }); + const agents = () => api.get(`${companyPath}/agents`); + const latestReply = async () => (await comments()).filter(comment => comment.authorAgentId === f.agent.id).at(-1)?.body ?? ""; + const output = (id: string, token: string) => readChatOutputDocument(api, id, token); + + try { + if (execution.task.id === "hire-delegate-reuse") { + const hireName = `Morgan Reviewer ${nonce}`; + await turn(`Hire exactly one teammate named ${hireName}, reporting to you, using your native runner, model, and available AI connection. Have that teammate write a concise launch checklist as a saved Paperclip document containing ${marker}. Put the work in one assigned task in ${project.name}. Link it here and let the teammate complete it.`, 2); + const first = (await tasks())[0]!; + expect(first).toBeTruthy(); + const firstOutput = await output(first.id, marker); + const account = f.aiConnection!; + const hired = assertChatHire({ agents: await agents(), leadId: f.agent.id, hireName, + connectionId: account.connectionId, binding: account.binding, taskIds: [first.id], tasks: await tasks(), runs: await allRuns() }); + expect((firstOutput as Document).createdByAgentId).toBe(hired.id); + await turn(`Have the existing ${hireName} review the checklist on ${first.identifier} and write a separate saved review document containing REVIEW${marker}. Create one review task in ${project.name}, assigned to that same teammate. Include the actual checklist in the handoff so they can review it. Preserve the original checklist and task.`, 4); + const observed = await tasks(); + const second = observed.find(task => task.id !== first.id)!; + expect(second).toBeTruthy(); + const reviewOutput = await output(second.id, `REVIEW${marker}`); + expect((reviewOutput as Document).createdByAgentId).toBe(hired.id); + expect(await api.get(`/api/issues/${first.id}/documents/${encodeURIComponent(firstOutput.key)}`)).toEqual(firstOutput); + await turn(`Give me a brief status update on ${first.identifier} and ${second.identifier}: identify their owner and recorded status. Do not create or change work.`, 5); + assertChatHire({ agents: await agents(), leadId: f.agent.id, hireName, hiredId: hired.id, + connectionId: account.connectionId, binding: account.binding, taskIds: [first.id, second.id], tasks: await tasks(), runs: await allRuns() }); + const reply = await latestReply(); + expect(reply).toContain(first.identifier!); expect(reply).toContain(second.identifier!); + expect(reply).toContain(hireName); expect(reply).toMatch(/done|completed/i); + await input.evidence("chat-hire-reuse.json", { agents: await agents(), tasks: await tasks(), firstOutput, reviewOutput, runs: await allRuns() }); + } else if (execution.task.id === "blocked-status-review") { + const blocker = `VENUE${marker}`, staleBlocker = `BUDGET${marker}`; + const source = await api.post(`${companyPath}/issues`, { + title: `Launch brief ${nonce}`, status: "blocked", projectId: project.id, + description: "The launch brief is waiting for its recorded blocker to be resolved.", + initialPlan: "The approved launch day is Tuesday.", + }); + const docResponse = await api.request.put(`/api/issues/${source.id}/documents/brief`, { + data: { title: "Launch brief", format: "markdown", body: "The launch brief schedules the launch for Wednesday." }, + }); + expect(docResponse.ok()).toBe(true); + const sourcePlan = await api.get(`/api/issues/${source.id}/documents/plan`); + const sourceBrief = await api.get(`/api/issues/${source.id}/documents/brief`); + await api.post(`/api/issues/${source.id}/comments`, { body: `Earlier blocker: ${staleBlocker}. Waiting for the budget.` }); + await api.post(`/api/issues/${source.id}/comments`, { body: `Budget is resolved. Current blocker: ${blocker}. Waiting for the venue confirmation. Keep this task blocked; no execution is active.` }); + const sourceBeforeStatus = await api.get(`/api/issues/${source.id}`); + await input.evidence("chat-status-review.json", { source: sourceBeforeStatus, sourcePlan, sourceBrief }); + await turn(`What is the actual current status of ${source.identifier}? Read its latest recorded blocker and execution state. Reply with only JSON containing issueIdentifier, status, currentBlockerLabel, and activeRunCount. You may include an explanation field. Just report; do not change it or create work.`, 1); + const sourceAfterStatus = await api.get(`/api/issues/${source.id}`); + assertGroundedChatStatus({ reply: await latestReply(), expectedIssueIdentifier: source.identifier!, blocker, + before: sourceBeforeStatus, after: sourceAfterStatus, taskIdsBefore: [source.id], + taskIdsAfter: (await tasks()).map(task => task.id), taskRuns: (await allRuns()).filter(run => run.contextSnapshot?.issueId === source.id) }); + const config = execution.profile.buildAgent({ environmentId: f.environment.id, environmentFixtureId: execution.environment.id, + workspacePath: input.workspacePath, secretRefs: f.secretRefs, executionId: nonce }); + const reviewer = await api.post(`${companyPath}/agents`, { ...config, name: `Riley Reviewer ${nonce}`, role: "engineer", reportsTo: f.agent.id }); + await turn(`Have ${reviewer.name} review the saved brief on ${source.identifier} against that task's saved plan, in one separate task in ${project.name}. Give the reviewer both source documents. They should save a JSON review document with planLaunchDay, briefLaunchDay, and a boolean consistent, then complete the review task. Preserve the source task and both documents.`, 3); + const review = (await tasks()).find(task => task.id !== source.id)!; + expect(review).toMatchObject({ status: "done", assigneeAgentId: reviewer.id, parentId: null, projectId: project.id }); + const document = await output(review.id, "planLaunchDay"); + assertChatSourceReview(document.body, { planLaunchDay: "Tuesday", briefLaunchDay: "Wednesday" }); + expect((document as Document).createdByAgentId).toBe(reviewer.id); + const reviewRuns = (await allRuns()).filter(run => run.contextSnapshot?.issueId === review.id); + expect(reviewRuns).toHaveLength(1); + expect(reviewRuns[0]).toMatchObject({ agentId: reviewer.id, status: "succeeded", runtimeMode: "native" }); + await turn(`Summarize ${review.identifier}'s recorded review result and tell me whether ${source.identifier} is still blocked. Read the saved evidence; do not change or create work.`, 4); + const reply = await latestReply(); + expect(reply).toContain("Tuesday"); expect(reply).toContain("Wednesday"); expect(reply).toContain(source.identifier!); expect(reply).toMatch(/blocked/i); + expect((await tasks()).map(task => task.id).sort()).toEqual([source.id, review.id].sort()); + expect(await api.get(`/api/issues/${source.id}/documents/plan`)).toEqual(sourcePlan); + expect(await api.get(`/api/issues/${source.id}/documents/brief`)).toEqual(sourceBrief); + expect(await api.get(`/api/issues/${source.id}`)).toMatchObject({ status: "blocked" }); + await input.evidence("chat-status-review.json", { source: sourceBeforeStatus, sourceAfterStatus, sourcePlan, sourceBrief, review, document, reviewRuns, reply }); + } else if (execution.task.id === "committed-send-retry") { + const interception = await dropChatSendAcknowledgement(page, marker); + try { + await sendChatMessage(page, `Create exactly one task titled Later ${nonce} in ${project.name}, assigned to yourself, in backlog. Save an initial plan containing ${marker}. Do not start it. Link the task here.`); + const sent = await interception.committed; + // The real agent completes the accepted request despite its lost HTTP ACK. + await idle(1); + const saved = (await tasks())[0]!; + expect(saved).toBeTruthy(); + const plan = await api.get(`/api/issues/${saved.id}/documents/plan`); + expect(plan.body).toContain(marker); + await interception.dispose(); + await input.restart(); + // Replay the exact interrupted public request, including its original key. + // This tests transport retries, not automatic replay of a failed provider turn. + const retried = await api.post(sent!.path, sent!.data); + expect(retried.id).toBe(sent!.commentId); + await page.goto(`/${f.company.issuePrefix}/chats/${f.agent.id}`, { waitUntil: "commit", timeout: 60_000 }); + await expect(page.getByTestId("task-chat-composer-input")).toBeVisible({ timeout: 60_000 }); + await turn(`What is the saved status of ${saved.identifier}? Just report it; do not create or change work.`, 2); + assertCommittedSendRetry({ commentId: sent!.commentId, clientRequestId: String(sent!.data.clientRequestId), + comments: await comments(), taskId: saved.id, tasks: await tasks(), runs: await allRuns(), chatId: context.issue().id }); + expect(await api.get(`/api/issues/${saved.id}/documents/plan`)).toEqual(plan); + expect(await latestReply()).toMatch(/backlog/i); + await input.evidence("chat-committed-send-retry.json", { commentId: sent!.commentId, task: saved, plan, runs: await allRuns() }); + } finally { + await interception.dispose(); + } + } else throw new Error(`Unsupported chat hardening case ${execution.task.id}`); + } finally { + // Preserve the records the matchers inspected even when an assertion fails. + // Failed API reads remain explicit evidence instead of replacing the failure. + const capture = async (name: string, load: () => Promise) => { + try { return { name, value: await load() }; } + catch (error) { return { name, error: String(error) }; } + }; + const state = await Promise.all([ + capture("agents", agents), + capture("runs", allRuns), + capture("chatComments", comments), + capture("tasks", async () => Promise.all((await tasks()).map(async task => ({ + task, + comments: await capture("comments", () => api.get(`/api/issues/${task.id}/comments?order=asc`)), + documents: await capture("documents", async () => { + const summaries = await api.get>(`/api/issues/${task.id}/documents`); + return Promise.all(summaries.map(document => capture(document.key, + () => api.get(`/api/issues/${task.id}/documents/${encodeURIComponent(document.key)}`)))); + }), + })))), + ]); + await input.evidence("chat-hardening-state.json", state); + } +} diff --git a/tests/runner-e2e/harness-env.ts b/tests/runner-e2e/harness-env.ts index d1e9cede374..6539f905b81 100644 --- a/tests/runner-e2e/harness-env.ts +++ b/tests/runner-e2e/harness-env.ts @@ -1,4 +1,5 @@ import path from "node:path"; +import { chatNeedsApiTools, isManagedHiringCase } from "./chat-cases.js"; import { CREDENTIAL_NAMES } from "./types.js"; import type { MatrixExecution } from "./types.js"; @@ -109,7 +110,7 @@ export function buildRunnerE2EProcessEnvironment( delete result.OPENCODE_ALLOW_ALL_MODELS; // Hiring needs the opt-in native API surface. Scope this to the explicit // manual hiring story; production and other suites retain their defaults. - if (executions.some((e) => e.suite.id === "everyday-workflows" && e.task.id === "hire-reuse")) { + if (executions.some((e) => isManagedHiringCase(e.suite.id, e.task.id) || chatNeedsApiTools(e.suite.id, e.task.id))) { result.PAPERCLIP_RUNNER_API_TOOLS_ENABLED = "true"; } if ( diff --git a/tests/runner-e2e/live-fixtures.test.ts b/tests/runner-e2e/live-fixtures.test.ts index 9c2eca9e15a..78cc912d619 100644 --- a/tests/runner-e2e/live-fixtures.test.ts +++ b/tests/runner-e2e/live-fixtures.test.ts @@ -4,13 +4,18 @@ import { runnerMatrix } from "./catalog.js"; import { setupLiveFixtures } from "./live-fixtures.js"; describe("live runner fixtures", () => { - it.each(["runner-codex", "runner-acpx-claude"])( - "gives %s hiring fixtures a personal managed account without env overrides", - async (profile) => { + it.each([ + ["runner-codex", "everyday-workflows", "hire-reuse"], + ["runner-acpx-claude", "everyday-workflows", "hire-reuse"], + ["runner-codex", "agent-chat-hardening", "hire-delegate-reuse"], + ["runner-acpx-claude", "agent-chat-hardening", "hire-delegate-reuse"], + ])( + "gives %s %s/%s a personal managed account without env overrides", + async (profile, suite, task) => { const execution = runnerMatrix.find( (e) => - e.suite.id === "everyday-workflows" && - e.task.id === "hire-reuse" && + e.suite.id === suite && + e.task.id === task && e.profile.id === profile && e.environment.id === "local", )!; diff --git a/tests/runner-e2e/live-fixtures.ts b/tests/runner-e2e/live-fixtures.ts index c7532a54a3c..edb960b851b 100644 --- a/tests/runner-e2e/live-fixtures.ts +++ b/tests/runner-e2e/live-fixtures.ts @@ -1,4 +1,5 @@ import path from "node:path"; +import { isManagedHiringCase } from "./chat-cases.js"; import { FixtureRegistry } from "./fixture-registry.js"; import type { RunnerApi } from "./api.js"; import type { @@ -212,9 +213,7 @@ export async function setupLiveFixtures(input: { }, }); - const managedHiring = - execution.suite.id === "everyday-workflows" && - execution.task.id === "hire-reuse"; + const managedHiring = isManagedHiringCase(execution.suite.id, execution.task.id); if (managedHiring) { registry.register({ id: "ai-connection", diff --git a/tests/runner-e2e/lost-send.spec.ts b/tests/runner-e2e/lost-send.spec.ts new file mode 100644 index 00000000000..0bde240508b --- /dev/null +++ b/tests/runner-e2e/lost-send.spec.ts @@ -0,0 +1,50 @@ +import { createServer } from "node:http"; +import { expect, test } from "@playwright/test"; +import { dropChatSendAcknowledgement } from "./lost-send.js"; + +test("drops the browser response only after the real server committed its comment", async ({ page }) => { + const requests: unknown[] = []; + const server = createServer(async (req, res) => { + if (req.method === "POST") { + let body = ""; + for await (const chunk of req) body += chunk; + requests.push(JSON.parse(body)); + res.writeHead(201, { "content-type": "application/json" }); + res.end(JSON.stringify({ id: "saved-comment" })); + return; + } + res.writeHead(200, { "content-type": "text/html" }); + res.end(``); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing fixture listener"); + const interception = await dropChatSendAcknowledgement(page, "PLAN123"); + try { + await page.goto(`http://127.0.0.1:${address.port}`); + await page.getByRole("button", { name: "Send" }).click(); + expect(await interception.committed).toEqual({ + path: "/api/issues/chat/comments", + data: { body: "Create PLAN123", clientRequestId: "original-key" }, + commentId: "saved-comment", + }); + expect(requests).toEqual([{ body: "Create PLAN123", clientRequestId: "original-key" }]); + await expect(page.locator("output")).toHaveText("response lost"); + await interception.dispose(); + await page.reload(); + await page.getByRole("button", { name: "Send" }).click(); + await expect(page.locator("output")).toHaveText("acknowledged"); + expect(requests).toHaveLength(2); + } finally { + await interception.dispose(); + server.closeAllConnections(); + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + } +}); diff --git a/tests/runner-e2e/lost-send.ts b/tests/runner-e2e/lost-send.ts new file mode 100644 index 00000000000..668e891bebf --- /dev/null +++ b/tests/runner-e2e/lost-send.ts @@ -0,0 +1,37 @@ +import type { Page, Route } from "@playwright/test"; + +export interface CommittedChatSend { + path: string; + data: Record; + commentId: string; +} + +/** Lose one real server acknowledgement, without keeping an expired browser request open. */ +export async function dropChatSendAcknowledgement(page: Page, marker: string) { + let resolve!: (value: CommittedChatSend) => void; + let reject!: (error: unknown) => void; + const committed = new Promise((yes, no) => { resolve = yes; reject = no; }); + // The route can fail before the caller finishes clicking Send. The caller + // still observes the original rejection when it awaits committed. + void committed.catch(() => undefined); + let claimed = false; + const pattern = "**/api/issues/*/comments"; + const handler = async (route: Route) => { + if (route.request().method() !== "POST" || claimed) return route.continue(); + const data = route.request().postDataJSON() as Record; + if (typeof data.body !== "string" || !data.body.includes(marker)) return route.continue(); + claimed = true; + try { + const response = await route.fetch(); + if (!response.ok()) throw new Error(`Intercepted chat send returned ${response.status()}`); + const comment = await response.json() as { id?: unknown }; + if (typeof comment.id !== "string") throw new Error("Intercepted chat send has no committed comment ID"); + await route.abort("connectionreset"); + resolve({ path: new URL(route.request().url()).pathname, data, commentId: comment.id }); + } catch (error) { + reject(error); + } + }; + await page.route(pattern, handler); + return { committed, dispose: () => page.unroute(pattern, handler) }; +} diff --git a/tests/runner-e2e/playwright-support.config.ts b/tests/runner-e2e/playwright-support.config.ts index c9905f3ecd2..517dcfa332c 100644 --- a/tests/runner-e2e/playwright-support.config.ts +++ b/tests/runner-e2e/playwright-support.config.ts @@ -3,7 +3,7 @@ import { defineConfig } from "@playwright/test"; /** Browser-only harness regressions: no Paperclip instance or provider credentials. */ export default defineConfig({ testDir: ".", - testMatch: "screenshot-readiness.spec.ts", + testMatch: ["screenshot-readiness.spec.ts", "lost-send.spec.ts"], workers: 1, retries: 0, timeout: 10_000,