mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(runner): resend an ACPX turn whose start was cut off before the prompt left (#20675)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A native ACPX run (for example Claude) sends `turn.start` to its Runner. The Runner asks the ACPX sidecar to start the turn, and the sidecar sends the prompt to the provider > - A start can fail without a definite answer. A host freeze can outlast the sidecar request deadline, or the sidecar or Runner can die during the start > - The Runner then saved the session `closed`. Every same-run resume failed with `ACPX session is not open`, and the task needed Retry > - The Runner cannot tell from its own state whether the provider has the prompt. Resending could duplicate it. A continuation could lose it > - This pull request adds a durable record that answers that question: a sidecar prompt-dispatch journal, written before the prompt reaches the provider > - The benefit is that a start cut off before the prompt left resends the prompt exactly once, with no Retry. Every case the record cannot prove keeps today's behavior ## Linked Issues or Issue Description **What happened?** Local restart-chaos wave `w-20261110T185315-s9107`, task t05 (claude, workload C1, fault TM2). A follow-up run reopened the parked Claude session and sent `turn.start`. The sidecar first starts Claude and loads the conversation; the prompt goes out only after that. After 24 s without a reply, the harness froze the server, Runner, sidecar and Claude (SIGSTOP) for 33 s. On thaw the Runner's 30 s sidecar deadline had passed. It reaped the sidecar and saved the session `closed`. Three same-run resumes failed with `PRP command turn.start failed: … ACPX session is not open`, recovery stopped with `native_session_retry_exhausted`, and a person had to click Retry. **Expected behavior** When a durable record proves the prompt never reached the provider, the run resends it once by itself. When the record cannot prove it, the run behaves as today. **Steps to reproduce** 1. On master, run `cargo test --manifest-path packages/paperclip-runner/runner/Cargo.toml -p paperclip-runner-core --test acpx_park_preparation cut_turn_start`. 2. `a_start_cut_off_before_the_prompt_left_reopens_and_sends_the_prompt_once` fails: the session is `closed` and the next start fails with `ACPX session is not open`. Refs #20507, #20645, #20674 ## What Changed - New file contract in `protocol/provider-schemas/acpx-sidecar.schema.json`: `turnDispatchJournalFile` (`paperclip-turn-dispatch.jsonl`) and `turnDispatchJournalRecord` (`received` / `dispatched`). The generator emits the file name and schema into TypeScript and Rust, so the pack contract digest covers them. - Sidecar (`turn-dispatch-journal.ts`, `codex-runtime-adapter.ts`, `runtime-host.ts`, `acpx-runtime-sidecar.ts`): when the runtime port starts a turn, before any provider work, it truncates the journal and writes `received`. ACPX calls `onAcpMessage("outbound")` synchronously before it writes a message to the provider. On the turn's first `session/prompt` request the port appends `dispatched` and fsyncs it there, so the record exists before the bytes leave. A failed dispatch write aborts the prompt write. A failed `received` write removes the file and runs the turn unjournaled. - Runner (`acpx_turn_dispatch.rs`, `acpx_provider_backend.rs`): the journal is read only after the provider lifetime fence proves that the sidecar and provider are gone. A lone complete `received` record for the turn proves the prompt never left. - A start that failed in flight saves the session `prepared` with no active turn and returns `ACPX turn start failed before its prompt reached the provider`. The run's next `turn.start` reopens the conversation and sends the prompt once. Until the proof settles, the saved state stays `turn_starting`, so a Runner killed during the proof goes through the restore check below. - A Runner killed while the session was `turn_starting` restores the session idle (`session.resumed`, no `turn.failed`). The same-run resume then starts the turn once instead of sending a continuation that would lose the prompt. - A `dispatched` record, another turn's record, a torn or malformed file, or no file keeps today's behavior (closed session, or lost-turn recovery). - Fake sidecar: writes the same journal and adds `hang-before-dispatch` / `hang-after-dispatch` faults. - Docs: `packages/paperclip-runner/docs/durable-recovery.md` ("ACPX turn starts cut off before the prompt left"). The design note is `doc/plans/2026-11-10-durable-turn-start-and-tool-call-records.md` (in #20674). ## Verification - Red on master, green here: `cargo test -p paperclip-runner-core --test acpx_park_preparation cut_turn_start` (4 tests: cut before dispatch reopens and sends once; cut after dispatch stays closed and never resends; Runner killed before dispatch restores idle; Runner killed after dispatch or with no journal reports the lost turn). The two "before" cases fail on master. - `cargo test -p paperclip-runner-core`: all pass. `cargo fmt --check` clean. - `pnpm --filter @paperclipai/paperclip-runner exec vitest run src/drivers/acpx/turn-dispatch-journal.test.ts src/drivers/acpx/codex-runtime-adapter.test.ts`: 121 passed. One test uses a real ACPX runtime and an ACP peer that refuses a prompt arriving before its `dispatched` record; it fails if the hook is removed. - `node scripts/generate-acpx-sidecar-contract.mjs --check`, `node --test scripts/contracts/contract-fingerprints.test.mjs`, and `tsc -p tsconfig.test.json` pass. - Time to first token, measured with the same open/write/fsync sequence on this Mac (APFS, load average ~11, 300 samples): `received` p50 0.24 ms, p95 0.59 ms; `dispatched` (with fsync) p50 4.7 ms, p95 11.5 ms. Against a Claude TTFT budget of 9.5–16.5 s p50 this is under 0.1%. No wait, round trip or file transfer is added. Recovery work (fence, journal read) runs only after a failed start. - Live restart-chaos wave and runner image: see the comments on this pull request. ## Risks - A false "not dispatched" proof would duplicate a prompt. It needs a lone, complete `received` record for this exact turn ID, read after the fence proves the sidecar is gone. `dispatched` is fsynced before the prompt write, and a failed write aborts the send. A failed `received` write removes the file, so it cannot leave a lone record. - Upgrade safety: saved provider state gains no field (its decoder denies unknown fields), and only lifecycles an older Runner already handles are written. An older Runner ignores the journal file. State saved by an older Runner has no journal, which reads as unknown and keeps today's behavior. A new Runner with an older sidecar never sees a `received` record, so no proof forms. - The changed sidecar schema changes the pack contract digest, so Daytona needs a new runner image. Older image packs are not reused. - A failed start now waits for the provider lifetime fence (at most 5 s, usually immediate) before it settles. This is only on a failure path. ## Model Used Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell, file editing) in Claude Code. Extended reasoning was used. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
DottaandClaude authored and GitHub committed
1 parent
b93df4ca38
commit
3fdcc2a28c
19 files changed
+956
-10
No files matched your search
@@ -53,4 +53,17 @@ This makes the gateway call idempotent on its durable input, which is the cleane
|
||||
|
||||
## Time to first token
|
||||
|
||||
Change 2 touches only recovery. Change 1 adds to every turn, before the prompt reaches the provider: one small unsynced file write when the turn starts, and one append plus `fsync` immediately before the prompt write. It adds no wait, no file transfer and no round trip. The measured cost is in the change 1 PR (target: well under 5 ms p95 on local disks, against a 9.5–16.5 s TTFT p50 budget). Recovery-time work (fence proof, journal read) runs only after a failure.
|
||||
Change 2 touches only recovery. Change 1 adds to every turn, before the prompt reaches the provider: one small unsynced file write when the turn starts, and one append plus `fsync` immediately before the prompt write. It adds no wait, no file transfer and no round trip. Measured with the same open/write/fsync sequence on a loaded macOS host (APFS, load average ~11, 300 samples): the `received` write p50 0.24 ms, p95 0.59 ms; the `dispatched` append and fsync p50 4.7 ms, p95 11.5 ms. Against a Claude TTFT budget of 9.5–16.5 s p50 this is under 0.1%. Linux overlay disks in Daytona were not measured. Recovery-time work (fence proof, journal read) runs only after a failure.
|
||||
|
||||
## Not covered: a remote Runner gone mid-turn with tools running
|
||||
|
||||
Staging class R, item 1 (`runner_remote_process_identity_unavailable`, mostly OpenCode on Daytona). A deploy stops the controller for 4–7 min while a turn runs tools. On recovery the remote Runner is gone (`process_not_running`, `runnerSettled: false`), or it is adopted and `session.snapshot` then times out. The turn is unsettled and tool effects are possible, so recovery refuses (`provider_turn_unsettled`) and Retry is needed. Retry works.
|
||||
|
||||
The question is whether a durable record can prove "no tool call was in flight when the Runner stopped" (every intent has an outcome), so recovery could continue from the checkpoint. **These two changes do not cover it.** What they give, and what is missing:
|
||||
|
||||
- **Paperclip semantic tools** (calls that reach the server's tool gateway). The intent is the controller's committed semantic input, keyed by call ID. The outcome is the server receipt or the journaled result command. Both live on the control plane, so they are readable without a live Runner. Change 2 makes `write_document` settle exactly once from its receipt. Other semantic tools keep their rules.
|
||||
- **Provider-native tools** (OpenCode's shell and file edits, Claude's built-in tools). These are the class R calls, and they have **no write-ahead intent record**. The Runner learns of a tool only from a provider event (`tool.execution.started`), which the provider emits after it decided to run the tool, and often after it started it. A missing `started` event therefore does not prove that no tool ran. A complete record needs a hook that runs before execution: a permission or pre-tool callback that the provider waits on (OpenCode plugins and ACP permission requests can serve), which journals the intent durably in the Runner's state before it answers.
|
||||
- **Readable without a live Runner.** The Runner's journal (`runner-state.json`, provider state) is in the sandbox. Commands already delivered stay on the controller, but events the Runner had not yet delivered are only in the sandbox. The server would need to read the sandbox files through the provider API (Daytona filesystem) after it proves the Runner and provider processes are stopped. It would also need a way to trust those bytes: today they carry no signature that binds them to the Runner's authenticated session. Process proof alone is not enough for an adopted Runner whose `session.snapshot` timed out; that case needs the same file read after a forced stop.
|
||||
- **The rule it would enable.** Every journaled intent has a matching outcome, the provider and Runner are proven stopped, and the last checkpoint covers the last outcome: recovery may resume from the checkpoint and send a same-conversation continuation. Anything else keeps Retry.
|
||||
|
||||
This is a follow-up, not part of these PRs.
|
||||
@@ -112,6 +112,47 @@ unconfirmed cleanup do not qualify. Error messages cannot authorize recovery.
|
||||
An arbitrary `closed` provider state stays closed. This path does not establish
|
||||
recovery for a Runner killed during an accepted turn.
|
||||
|
||||
## ACPX turn starts cut off before the prompt left
|
||||
|
||||
A turn start can fail without a definite answer: the sidecar's reply is lost or
|
||||
late (for example, a host freeze outlasts the request deadline), or the sidecar
|
||||
or Runner dies while the start is in flight. Resending the prompt could
|
||||
duplicate it, and a continuation could lose it.
|
||||
|
||||
The sidecar's runtime port keeps a prompt-dispatch journal,
|
||||
`paperclip-turn-dispatch.jsonl`, in the session's private runtime directory
|
||||
(`turnDispatchJournalFile` and `turnDispatchJournalRecord` in the sidecar
|
||||
schema). When it starts a turn, before any provider work, it truncates the file
|
||||
and writes a `received` record. Immediately before ACPX writes the turn's first
|
||||
`session/prompt` request to the provider, it appends a `dispatched` record and
|
||||
fsyncs it. ACPX calls the host's `onAcpMessage("outbound")` hook synchronously
|
||||
before it writes the bytes, so a failed journal write aborts the prompt write.
|
||||
|
||||
The Runner reads the journal only after it acquires the provider lifetime fence,
|
||||
which proves that the sidecar and the provider are gone.
|
||||
|
||||
- A lone, complete `received` record for the turn proves that the prompt never
|
||||
reached the provider. A start that failed in flight saves the session
|
||||
`prepared`, with no active turn, and fails the command with
|
||||
`ACPX turn start failed before its prompt reached the provider`. The run's
|
||||
next `turn.start` reopens the conversation and sends the prompt once. A
|
||||
Runner that died while the session was `turn_starting` restores the session
|
||||
idle, emits `session.resumed` and no `turn.failed`, and the same-run resume
|
||||
starts the turn once.
|
||||
- A `dispatched` record, a record of another turn, a torn or malformed file, or
|
||||
no file (an older sidecar) proves nothing. The earlier behavior applies
|
||||
unchanged: a failed start closes the session, and a killed Runner uses
|
||||
lost-turn recovery.
|
||||
|
||||
Until a proof settles it, the saved session stays `turn_starting`, so a Runner
|
||||
killed during the proof restores through the same check. The cut-off command is
|
||||
never executed again. Saved state gains no field, so an older Runner reading it
|
||||
mid-deploy sees only lifecycles it already handles.
|
||||
|
||||
On the normal path the journal costs one small unsynced write when the turn
|
||||
starts and one append and fsync just before the prompt is written. It adds no
|
||||
wait, round trip or file transfer.
|
||||
|
||||
## ACPX cancellation without a reply
|
||||
|
||||
The Runner serves controller commands one at a time. It therefore waits at most
|
||||
|
||||
@@ -9,6 +9,21 @@
|
||||
],
|
||||
"$defs": {
|
||||
"protocolVersion": { "const": 2 },
|
||||
"turnDispatchJournalFile": {
|
||||
"description": "File name of the sidecar's prompt-dispatch journal in the session runtime directory. It holds turnDispatchJournalRecord lines for the latest turn only.",
|
||||
"const": "paperclip-turn-dispatch.jsonl"
|
||||
},
|
||||
"turnDispatchJournalRecord": {
|
||||
"description": "One journal line. The runtime port truncates the file and writes `received` when it starts the turn, before any provider work. It appends `dispatched` and fsyncs it immediately before the turn's first session/prompt request is written to the provider. The Runner reads the journal only after it proves that the sidecar and provider are gone: a lone `received` line proves the prompt never reached the provider.",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"schema": { "const": "paperclip.acpx-turn-dispatch.v1" },
|
||||
"turnId": { "type": "string", "minLength": 1 },
|
||||
"phase": { "enum": ["received", "dispatched"] }
|
||||
},
|
||||
"required": ["schema", "turnId", "phase"],
|
||||
"additionalProperties": false
|
||||
},
|
||||
"command": {
|
||||
"enum": [
|
||||
"initialize",
|
||||
|
||||
@@ -22,6 +22,7 @@ use crate::acpx_provider_session::{
|
||||
PiThinkingLevel,
|
||||
};
|
||||
use crate::acpx_sidecar_transport::AcpxSidecarTransportConfig;
|
||||
use crate::acpx_turn_dispatch::{read_turn_dispatch_journal, TurnDispatchEvidence};
|
||||
#[cfg(test)]
|
||||
use crate::durable::QualifiedLaunchArtifact;
|
||||
use crate::durable::{
|
||||
@@ -1299,6 +1300,66 @@ impl AcpxCommandExecutor {
|
||||
self.save_state()
|
||||
}
|
||||
|
||||
/// A turn start that failed without a definite answer (a lost or late
|
||||
/// reply, an exited or frozen sidecar) leaves the prompt's fate unknown:
|
||||
/// resending could duplicate it and continuing could lose it. Once the
|
||||
/// sidecar and provider are proven gone, the sidecar's dispatch journal
|
||||
/// can prove that the prompt never left. Then the session is saved
|
||||
/// reopenable with no active turn, so the run's next `turn.start` sends
|
||||
/// the prompt once. Returns false, changing nothing, without that proof.
|
||||
fn reopen_after_undispatched_turn_start(
|
||||
&mut self,
|
||||
turn_id: &str,
|
||||
) -> Result<bool, DurableRunnerError> {
|
||||
let session_identity = self
|
||||
.session
|
||||
.as_ref()
|
||||
.map(|session| session.identity().clone());
|
||||
let Some(state) = self.state.as_ref() else {
|
||||
return Ok(false);
|
||||
};
|
||||
let Some(identity) = state.identity.clone() else {
|
||||
return Ok(false);
|
||||
};
|
||||
if session_identity.as_ref() != Some(&identity)
|
||||
|| state.lifecycle != "turn_starting"
|
||||
|| state.active_turn_id.as_deref() != Some(turn_id)
|
||||
|| state.provider_exit_unconfirmed
|
||||
|| !state.pending_runtime_requests.is_empty()
|
||||
|| state.semantic_result.is_some()
|
||||
|| !state
|
||||
.pending_events
|
||||
.iter()
|
||||
.all(|event| event.event_type == "session.resumed")
|
||||
|| !(state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection))
|
||||
{
|
||||
return Ok(false);
|
||||
}
|
||||
// The saved state still says `turn_starting` until a proof settles it,
|
||||
// so a Runner killed during this wait restores through the same check.
|
||||
self.session = None;
|
||||
let Ok(_provider_lifetime_fence) =
|
||||
await_provider_lifetime_fence(identity.provider_lifetime_fence_candidates)
|
||||
else {
|
||||
return Ok(false);
|
||||
};
|
||||
let runtime_directory = state.descriptor.runtime_directory.clone();
|
||||
if read_turn_dispatch_journal(&runtime_directory, turn_id)
|
||||
!= TurnDispatchEvidence::NotDispatched
|
||||
{
|
||||
return Ok(false);
|
||||
}
|
||||
let state = self
|
||||
.state
|
||||
.as_mut()
|
||||
.expect("ACPX state remains available after an undispatched turn start");
|
||||
state.lifecycle = "prepared".to_owned();
|
||||
state.active_turn_id = None;
|
||||
self.context.provider_turn_id = None;
|
||||
self.save_state()?;
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// Restores the conversation of a turn that was active when the runner
|
||||
/// died, without inventing its outcome. The original provider lifetime
|
||||
/// must have lost its fence quorum first, so no second provider can act on
|
||||
@@ -1326,11 +1387,29 @@ impl AcpxCommandExecutor {
|
||||
{
|
||||
return Ok(false);
|
||||
}
|
||||
match await_provider_lifetime_fence(identity.provider_lifetime_fence_candidates) {
|
||||
// Release the quorum so the restored provider can inherit it.
|
||||
Ok(listeners) => drop(listeners),
|
||||
Err(_) => return Ok(false),
|
||||
}
|
||||
// A Runner killed while the start was in flight saved `turn_starting`.
|
||||
// When the sidecar's journal proves the prompt never left, no
|
||||
// provider turn exists: restore the session idle so the run's next
|
||||
// turn.start sends the prompt once, instead of a continuation that
|
||||
// would lose it.
|
||||
let start_turn_id = (state.lifecycle == "turn_starting")
|
||||
.then(|| previous_turn.clone())
|
||||
.flatten();
|
||||
let runtime_directory = state.descriptor.runtime_directory.clone();
|
||||
let undispatched =
|
||||
match await_provider_lifetime_fence(identity.provider_lifetime_fence_candidates) {
|
||||
// Read the journal while the quorum proves the sidecar is gone,
|
||||
// then release it so the restored provider can inherit it.
|
||||
Ok(listeners) => {
|
||||
let undispatched = start_turn_id.is_some_and(|turn_id| {
|
||||
read_turn_dispatch_journal(&runtime_directory, &turn_id)
|
||||
== TurnDispatchEvidence::NotDispatched
|
||||
});
|
||||
drop(listeners);
|
||||
undispatched
|
||||
}
|
||||
Err(_) => return Ok(false),
|
||||
};
|
||||
let Ok(session) = self.start_session(true) else {
|
||||
return Ok(false);
|
||||
};
|
||||
@@ -1353,6 +1432,12 @@ impl AcpxCommandExecutor {
|
||||
priority: EventPriority::P0,
|
||||
payload: session_event_payload(&state.descriptor, &identity, process_id, turn_controls),
|
||||
})?;
|
||||
if undispatched {
|
||||
self.recovery_pending_effects = None;
|
||||
self.session = Some(session);
|
||||
self.save_state()?;
|
||||
return Ok(true);
|
||||
}
|
||||
state.push(NormalizedProviderEvent {
|
||||
event_type: "turn.failed".to_owned(),
|
||||
priority: EventPriority::P0,
|
||||
@@ -2103,6 +2188,13 @@ impl AcpxCommandExecutor {
|
||||
})
|
||||
})
|
||||
.cloned();
|
||||
if rejected_checkpoint.is_none()
|
||||
&& self.reopen_after_undispatched_turn_start(&provider_turn_id)?
|
||||
{
|
||||
return Err(DurableRunnerError::invalid(format!(
|
||||
"ACPX turn start failed before its prompt reached the provider: {error}"
|
||||
)));
|
||||
}
|
||||
let state = self
|
||||
.state
|
||||
.as_mut()
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
//! Reads the ACPX sidecar's prompt-dispatch journal.
|
||||
//!
|
||||
//! The sidecar's runtime port writes `received` for a turn before any provider
|
||||
//! work and appends a synced `dispatched` immediately before the turn's first
|
||||
//! `session/prompt` request reaches the provider (see the TypeScript
|
||||
//! `turn-dispatch-journal.ts` and `turnDispatchJournalRecord` in the sidecar
|
||||
//! schema). Callers read it only after they prove that the sidecar and the
|
||||
//! provider are gone, so neither can still write or send.
|
||||
|
||||
use std::io::Read;
|
||||
use std::path::Path;
|
||||
|
||||
use serde::Deserialize;
|
||||
|
||||
use crate::durable::open_private_regular_file;
|
||||
use crate::generated_acpx_sidecar_contract::{
|
||||
GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE, GENERATED_ACPX_TURN_DISPATCH_SCHEMA,
|
||||
};
|
||||
|
||||
// Two records of one bounded turn ID fit easily; anything larger is not ours.
|
||||
const MAX_JOURNAL_BYTES: u64 = 4 * 1024;
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub(crate) enum TurnDispatchEvidence {
|
||||
/// The sidecar started this turn and never wrote its prompt to the provider.
|
||||
NotDispatched,
|
||||
/// The prompt may have reached the provider.
|
||||
Dispatched,
|
||||
/// No usable record for this turn: an older sidecar, another turn, or a
|
||||
/// damaged file. Recovery must not assume either outcome.
|
||||
Unknown,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct Record {
|
||||
schema: String,
|
||||
#[serde(rename = "turnId")]
|
||||
turn_id: String,
|
||||
phase: String,
|
||||
}
|
||||
|
||||
pub(crate) fn read_turn_dispatch_journal(
|
||||
runtime_directory: &Path,
|
||||
turn_id: &str,
|
||||
) -> TurnDispatchEvidence {
|
||||
let path = runtime_directory.join(GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE);
|
||||
let Ok(file) = open_private_regular_file(&path) else {
|
||||
return TurnDispatchEvidence::Unknown;
|
||||
};
|
||||
let mut bytes = Vec::new();
|
||||
if file
|
||||
.take(MAX_JOURNAL_BYTES + 1)
|
||||
.read_to_end(&mut bytes)
|
||||
.is_err()
|
||||
|| bytes.len() as u64 > MAX_JOURNAL_BYTES
|
||||
{
|
||||
return TurnDispatchEvidence::Unknown;
|
||||
}
|
||||
classify(&bytes, turn_id)
|
||||
}
|
||||
|
||||
fn classify(bytes: &[u8], turn_id: &str) -> TurnDispatchEvidence {
|
||||
// Every record ends with a newline; a torn tail proves nothing.
|
||||
let Some(body) = bytes.strip_suffix(b"\n") else {
|
||||
return TurnDispatchEvidence::Unknown;
|
||||
};
|
||||
let mut phases = Vec::new();
|
||||
for line in body.split(|byte| *byte == b'\n') {
|
||||
let Ok(record) = serde_json::from_slice::<Record>(line) else {
|
||||
return TurnDispatchEvidence::Unknown;
|
||||
};
|
||||
if record.schema != GENERATED_ACPX_TURN_DISPATCH_SCHEMA || record.turn_id != turn_id {
|
||||
return TurnDispatchEvidence::Unknown;
|
||||
}
|
||||
phases.push(record.phase);
|
||||
}
|
||||
match phases.as_slice() {
|
||||
[received] if received == "received" => TurnDispatchEvidence::NotDispatched,
|
||||
[received, rest @ ..]
|
||||
if received == "received"
|
||||
&& !rest.is_empty()
|
||||
&& rest.iter().all(|phase| phase == "dispatched") =>
|
||||
{
|
||||
TurnDispatchEvidence::Dispatched
|
||||
}
|
||||
_ => TurnDispatchEvidence::Unknown,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn line(turn: &str, phase: &str) -> String {
|
||||
format!(
|
||||
"{{\"schema\":\"{GENERATED_ACPX_TURN_DISPATCH_SCHEMA}\",\"turnId\":\"{turn}\",\"phase\":\"{phase}\"}}\n"
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn only_a_lone_received_record_of_this_turn_proves_the_prompt_never_left() {
|
||||
let received = line("turn-1", "received");
|
||||
let dispatched = format!("{received}{}", line("turn-1", "dispatched"));
|
||||
assert_eq!(
|
||||
classify(received.as_bytes(), "turn-1"),
|
||||
TurnDispatchEvidence::NotDispatched
|
||||
);
|
||||
assert_eq!(
|
||||
classify(dispatched.as_bytes(), "turn-1"),
|
||||
TurnDispatchEvidence::Dispatched
|
||||
);
|
||||
for unknown in [
|
||||
String::new(),
|
||||
received.trim_end().to_owned(),
|
||||
line("turn-0", "received"),
|
||||
line("turn-1", "dispatched"),
|
||||
format!("{received}{}", line("turn-0", "dispatched")),
|
||||
format!("{received}{{\"schema\":"),
|
||||
format!("{received}{received}"),
|
||||
received.replace(
|
||||
"paperclip.acpx-turn-dispatch.v1",
|
||||
"paperclip.acpx-turn-dispatch.v0",
|
||||
),
|
||||
received.replace("}\n", ",\"extra\":true}\n"),
|
||||
] {
|
||||
assert_eq!(
|
||||
classify(unknown.as_bytes(), "turn-1"),
|
||||
TurnDispatchEvidence::Unknown,
|
||||
"{unknown:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,10 @@
|
||||
use std::io::{self, BufRead, Write};
|
||||
use std::time::Duration;
|
||||
|
||||
use paperclip_runner_core::generated_acpx_sidecar_contract::GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION;
|
||||
use paperclip_runner_core::generated_acpx_sidecar_contract::{
|
||||
GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE,
|
||||
GENERATED_ACPX_TURN_DISPATCH_SCHEMA,
|
||||
};
|
||||
use serde_json::{json, Value};
|
||||
|
||||
const PROJECTED_INPUT_PROVIDER_ID: &str = "input / réquest";
|
||||
@@ -69,6 +72,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let mut source_run = String::new();
|
||||
let mut source_pending_completion: Option<String> = None;
|
||||
let mut source_goal_revision = 0_u64;
|
||||
let mut runtime_directory: Option<std::path::PathBuf> = None;
|
||||
// `--cancel-reply lost|late`: the provider stops its prompt, but the reply
|
||||
// to turn.cancel is dropped, or written only before the next request.
|
||||
let cancel_reply = args
|
||||
@@ -252,6 +256,47 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
response = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":id,"ok":false,
|
||||
"error":{"code":"temporarily_unavailable","message":"Optional read is temporarily unavailable.","retryable":true}});
|
||||
}
|
||||
if command == "session.open" {
|
||||
runtime_directory = request["params"]["runtimeDirectory"]
|
||||
.as_str()
|
||||
.map(std::path::PathBuf::from);
|
||||
}
|
||||
// Like the real runtime port: journal the turn, then its prompt
|
||||
// dispatch, before the provider can see the prompt.
|
||||
// `hang-before-dispatch` / `hang-after-dispatch` freeze the start
|
||||
// on either side of that boundary without replying.
|
||||
if command == "turn.start" {
|
||||
if let Some(directory) = runtime_directory.as_ref() {
|
||||
let turn = request["params"]["turnId"].as_str().unwrap();
|
||||
let path = directory.join(GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE);
|
||||
let record = |phase: &str| {
|
||||
format!(
|
||||
"{}\n",
|
||||
json!({"schema":GENERATED_ACPX_TURN_DISPATCH_SCHEMA,"turnId":turn,"phase":phase})
|
||||
)
|
||||
};
|
||||
let mut options = std::fs::OpenOptions::new();
|
||||
options.write(true).create(true).truncate(true);
|
||||
#[cfg(unix)]
|
||||
std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
|
||||
options
|
||||
.open(&path)?
|
||||
.write_all(record("received").as_bytes())?;
|
||||
if fault == "hang-before-dispatch" {
|
||||
loop {
|
||||
std::thread::sleep(std::time::Duration::from_secs(60));
|
||||
}
|
||||
}
|
||||
let mut journal = std::fs::OpenOptions::new().append(true).open(&path)?;
|
||||
journal.write_all(record("dispatched").as_bytes())?;
|
||||
// The runtime port fsyncs here; the fixture skips it to stay fast.
|
||||
if fault == "hang-after-dispatch" {
|
||||
loop {
|
||||
std::thread::sleep(std::time::Duration::from_secs(60));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if command == "session.open" && fault == "slow-open" {
|
||||
std::thread::sleep(std::time::Duration::from_secs(10));
|
||||
}
|
||||
|
||||
+2
@@ -1,6 +1,8 @@
|
||||
// Generated by scripts/generate-acpx-sidecar-contract.mjs. Do not edit.
|
||||
|
||||
pub const GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION: u64 = 2;
|
||||
pub const GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE: &str = "paperclip-turn-dispatch.jsonl";
|
||||
pub const GENERATED_ACPX_TURN_DISPATCH_SCHEMA: &str = "paperclip.acpx-turn-dispatch.v1";
|
||||
|
||||
pub const GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE: &[(&str, &[&str])] = &[
|
||||
("edit", &["edit", "write", "patch"]),
|
||||
|
||||
@@ -9,6 +9,7 @@ pub mod acpx_provider_checkpoint;
|
||||
pub mod acpx_provider_session;
|
||||
pub mod acpx_provider_state;
|
||||
pub mod acpx_sidecar_transport;
|
||||
mod acpx_turn_dispatch;
|
||||
pub mod aws_agentcore_provider;
|
||||
pub mod claude_managed_provider;
|
||||
pub mod codex_provider;
|
||||
|
||||
@@ -17,6 +17,9 @@ use sha2::{Digest, Sha256};
|
||||
#[path = "support/acpx_turn_end_restart.rs"]
|
||||
mod turn_end_restart;
|
||||
|
||||
#[path = "support/acpx_cut_turn_start.rs"]
|
||||
mod cut_turn_start;
|
||||
|
||||
struct Fixture(PathBuf);
|
||||
impl Drop for Fixture {
|
||||
fn drop(&mut self) {
|
||||
|
||||
+204
@@ -0,0 +1,204 @@
|
||||
//! A `turn.start` cut off between the Runner's request and the sidecar's
|
||||
//! reply (a frozen or killed sidecar, a timeout). The sidecar's dispatch
|
||||
//! journal decides whether the prompt could have reached the provider.
|
||||
|
||||
use super::*;
|
||||
|
||||
const JOURNAL: &str = "paperclip-turn-dispatch.jsonl";
|
||||
|
||||
fn journal_phases(harness: &Harness) -> Vec<String> {
|
||||
fs::read_to_string(harness.fixture.0.join(JOURNAL))
|
||||
.unwrap_or_default()
|
||||
.lines()
|
||||
.map(|line| {
|
||||
serde_json::from_str::<Value>(line).unwrap()["phase"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_owned()
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn kill(pid: &Value) {
|
||||
std::process::Command::new("kill")
|
||||
.args(["-9", &pid.to_string()])
|
||||
.status()
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// Starts `turn` against a sidecar that freezes at `fault`, then kills the
|
||||
/// sidecar once the journal shows `phases`, and returns the start's error.
|
||||
fn cut_turn_start(harness: &mut Harness, turn: &str, fault: &str, phases: &[&str]) -> String {
|
||||
harness.fixture.fault(fault);
|
||||
let sidecar = harness.fixture.process_starts().last().unwrap()["processStart"].clone();
|
||||
let root = harness.fixture.0.clone();
|
||||
let expected = phases
|
||||
.iter()
|
||||
.map(|phase| phase.to_string())
|
||||
.collect::<Vec<_>>();
|
||||
let killer = std::thread::spawn(move || {
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
loop {
|
||||
let seen = fs::read_to_string(root.join(JOURNAL))
|
||||
.unwrap_or_default()
|
||||
.lines()
|
||||
.map(|line| {
|
||||
serde_json::from_str::<Value>(line).unwrap()["phase"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_owned()
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
if seen == expected {
|
||||
break;
|
||||
}
|
||||
assert!(
|
||||
Instant::now() < deadline,
|
||||
"the sidecar never reached {expected:?}"
|
||||
);
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
kill(&sidecar);
|
||||
});
|
||||
harness.sequence += 1;
|
||||
let error = harness
|
||||
.executor
|
||||
.execute(&command(
|
||||
harness.sequence,
|
||||
"turn.start",
|
||||
json!({"turnId": turn, "text": "Write the brief."}),
|
||||
))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
killer.join().unwrap();
|
||||
harness.fixture.fault("");
|
||||
error
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_start_cut_off_before_the_prompt_left_reopens_and_sends_the_prompt_once() {
|
||||
let mut harness = Harness::new();
|
||||
let error = cut_turn_start(&mut harness, "cut", "hang-before-dispatch", &["received"]);
|
||||
assert!(
|
||||
error.contains("before its prompt reached the provider"),
|
||||
"{error}"
|
||||
);
|
||||
let saved = harness.fixture.state();
|
||||
assert_eq!(saved["lifecycle"], "prepared", "{saved}");
|
||||
assert_eq!(saved["activeTurnId"], Value::Null);
|
||||
assert_eq!(saved["providerExitUnconfirmed"], false);
|
||||
// The run's next start reopens the conversation and delivers its prompt.
|
||||
let before = harness.fixture.journal().len();
|
||||
let events = harness.complete_turn("resent");
|
||||
assert!(
|
||||
events.iter().all(|event| event["type"] != "turn.failed"),
|
||||
"{events:?}"
|
||||
);
|
||||
let starts = harness.fixture.journal()[before..]
|
||||
.iter()
|
||||
.filter(|line| line["command"] == "turn.start")
|
||||
.map(|line| line["params"]["turnId"].clone())
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(starts, [json!("resent")]);
|
||||
assert_eq!(journal_phases(&harness), ["received", "dispatched"]);
|
||||
harness.executor.shutdown().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_start_cut_off_after_the_prompt_left_stays_closed_and_never_resends() {
|
||||
let mut harness = Harness::new();
|
||||
let error = cut_turn_start(
|
||||
&mut harness,
|
||||
"cut",
|
||||
"hang-after-dispatch",
|
||||
&["received", "dispatched"],
|
||||
);
|
||||
assert!(error.contains("failed closed"), "{error}");
|
||||
assert_eq!(harness.fixture.state()["lifecycle"], "closed");
|
||||
let starts = harness.fixture.process_starts().len();
|
||||
harness.sequence += 1;
|
||||
let refused = harness
|
||||
.executor
|
||||
.execute(&command(
|
||||
harness.sequence,
|
||||
"turn.start",
|
||||
json!({"turnId": "resent", "text": "Write the brief."}),
|
||||
))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert_eq!(refused, "ACPX session is not open");
|
||||
assert_eq!(harness.fixture.process_starts().len(), starts);
|
||||
}
|
||||
|
||||
/// The Runner itself dies while the session is saved `turn_starting`.
|
||||
fn restore_cut_runner(harness: &mut Harness, phases: &[&str]) -> Vec<Value> {
|
||||
let mut saved = harness.fixture.state();
|
||||
harness.executor.shutdown().unwrap();
|
||||
saved["lifecycle"] = json!("turn_starting");
|
||||
saved["activeTurnId"] = json!("cut");
|
||||
fs::write(
|
||||
harness.fixture.0.join("acpx-provider-state.json"),
|
||||
serde_json::to_vec_pretty(&saved).unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
let journal = phases
|
||||
.iter()
|
||||
.map(|phase| {
|
||||
format!(
|
||||
"{}\n",
|
||||
json!({"schema":"paperclip.acpx-turn-dispatch.v1","turnId":"cut","phase":phase})
|
||||
)
|
||||
})
|
||||
.collect::<String>();
|
||||
let path = harness.fixture.0.join(JOURNAL);
|
||||
let _ = fs::remove_file(&path);
|
||||
if !phases.is_empty() {
|
||||
fs::write(&path, journal).unwrap();
|
||||
fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap();
|
||||
}
|
||||
harness.executor =
|
||||
NativeProviderCommandExecutor::with_runner_config(&harness.fixture.0, &harness.config);
|
||||
harness.sequence += 1;
|
||||
let snapshot = harness
|
||||
.executor
|
||||
.execute(&command(harness.sequence, "session.snapshot", json!({})))
|
||||
.unwrap();
|
||||
let mut events = snapshot
|
||||
.events
|
||||
.iter()
|
||||
.map(|(kind, _, payload)| json!({"type":kind,"payload":payload}))
|
||||
.collect::<Vec<_>>();
|
||||
events.extend(harness.drain());
|
||||
events
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_runner_killed_before_the_prompt_left_restores_an_idle_session() {
|
||||
let mut harness = Harness::new();
|
||||
let events = restore_cut_runner(&mut harness, &["received"]);
|
||||
let kinds = events
|
||||
.iter()
|
||||
.map(|event| event["type"].clone())
|
||||
.collect::<Vec<_>>();
|
||||
assert!(kinds.contains(&json!("session.resumed")), "{events:?}");
|
||||
assert!(!kinds.contains(&json!("turn.failed")), "{events:?}");
|
||||
let saved = harness.fixture.state();
|
||||
assert_eq!(saved["lifecycle"], "session_open");
|
||||
assert_eq!(saved["activeTurnId"], Value::Null);
|
||||
harness.complete_turn("resent");
|
||||
harness.executor.shutdown().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_runner_killed_after_the_prompt_left_or_without_a_journal_reports_the_lost_turn() {
|
||||
for phases in [&["received", "dispatched"][..], &[][..]] {
|
||||
let mut harness = Harness::new();
|
||||
let events = restore_cut_runner(&mut harness, phases);
|
||||
let lost = events
|
||||
.iter()
|
||||
.find(|event| event["type"] == "turn.failed")
|
||||
.unwrap_or_else(|| panic!("{phases:?}: {events:?}"));
|
||||
assert_eq!(lost["payload"]["code"], "provider_turn_lost_on_restore");
|
||||
harness.executor.shutdown().unwrap();
|
||||
}
|
||||
}
|
||||
@@ -33,6 +33,21 @@ const toolOperationPrecedence = [
|
||||
["list", ["list", "glob"]],
|
||||
];
|
||||
const protocolVersion = readAcpxSidecarProtocolVersion(schema);
|
||||
// The prompt-dispatch journal is a file contract between the sidecar (writer)
|
||||
// and the Runner (reader after the sidecar is proven gone).
|
||||
const turnDispatchJournalFile = schema.$defs.turnDispatchJournalFile?.const;
|
||||
const turnDispatchRecord = schema.$defs.turnDispatchJournalRecord;
|
||||
const turnDispatchSchema = turnDispatchRecord?.properties?.schema?.const;
|
||||
const turnDispatchPhases = turnDispatchRecord?.properties?.phase?.enum;
|
||||
if (
|
||||
typeof turnDispatchJournalFile !== "string" ||
|
||||
!/^[a-z0-9.-]+$/.test(turnDispatchJournalFile) ||
|
||||
typeof turnDispatchSchema !== "string" ||
|
||||
JSON.stringify(turnDispatchPhases) !== JSON.stringify(["received", "dispatched"]) ||
|
||||
turnDispatchRecord?.additionalProperties !== false
|
||||
) {
|
||||
throw new Error("ACPX turn dispatch journal contract must declare its file, schema and closed phases");
|
||||
}
|
||||
const quote = (value) => JSON.stringify(value);
|
||||
const rustVariant = (value) =>
|
||||
value
|
||||
@@ -40,8 +55,8 @@ const rustVariant = (value) =>
|
||||
.map((part) => part[0].toUpperCase() + part.slice(1))
|
||||
.join("");
|
||||
|
||||
const typescript = `// Generated by scripts/generate-acpx-sidecar-contract.mjs. Do not edit.\n\nexport const GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION = ${protocolVersion} as const;\nexport const GENERATED_ACPX_SIDECAR_COMMANDS = [\n${commands.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\nexport type GeneratedAcpxSidecarCommand =\n (typeof GENERATED_ACPX_SIDECAR_COMMANDS)[number];\n\nexport const GENERATED_ACPX_SIDECAR_EVENT_TYPES = [\n${events.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\nexport type GeneratedAcpxSidecarEventType =\n (typeof GENERATED_ACPX_SIDECAR_EVENT_TYPES)[number];\n\nexport const GENERATED_ACPX_SIDECAR_OPEN_PARAMS = [\n${openParams.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\n\nexport const GENERATED_ACPX_SIDECAR_PROVIDER_POLICY_FIELDS = [\n${providerPolicyFields.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\n\nexport type GeneratedAcpxToolOperation =\n | "read"\n | "search"\n | "list"\n | "edit"\n | "execute"\n | "unknown";\n\nexport const GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE = [\n${toolOperationPrecedence.map(([operation, tokens]) => ` { operation: ${quote(operation)}, tokens: [${tokens.map(quote).join(", ")}] },`).join("\n")}\n] as const;\n\nexport function classifyGeneratedAcpxToolOperation(\n toolKind: unknown,\n toolTitle: unknown,\n): GeneratedAcpxToolOperation {\n const candidate =\n typeof toolKind === "string" && toolKind\n ? toolKind\n : typeof toolTitle === "string"\n ? toolTitle\n : "";\n const normalized = candidate.slice(0, 240).toLowerCase();\n for (const { operation, tokens } of GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE) {\n if (tokens.some((token) => normalized.includes(token))) return operation;\n }\n return normalized ? "execute" : "unknown";\n}\n`;
|
||||
const rust = `// Generated by scripts/generate-acpx-sidecar-contract.mjs. Do not edit.\n\npub const GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION: u64 = ${protocolVersion};\n\npub const GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE: &[(&str, &[&str])] = &[\n${toolOperationPrecedence.map(([operation, tokens]) => ` (${quote(operation)}, &[${tokens.map(quote).join(", ")}]),`).join("\n")}\n];\n\npub fn classify_generated_acpx_tool_operation(kind: &str, title: &str) -> &'static str {\n let kind = kind.to_ascii_lowercase();\n let title = title.to_ascii_lowercase();\n let candidate = if kind.is_empty() { &title } else { &kind };\n for (operation, tokens) in GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE {\n if tokens.iter().any(|token| candidate.contains(token)) {\n return operation;\n }\n }\n if candidate.is_empty() {\n "unknown"\n } else {\n "execute"\n }\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub enum GeneratedAcpxSidecarCommand {\n${commands.map((value) => ` ${rustVariant(value)},`).join("\n")}\n}\n\nimpl GeneratedAcpxSidecarCommand {\n pub const fn as_str(self) -> &'static str {\n match self {\n${commands.map((value) => ` Self::${rustVariant(value)} => ${quote(value)},`).join("\n")}\n }\n }\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Deserialize)]\npub enum GeneratedAcpxSidecarEventType {\n${events.map((value) => ` #[serde(rename = ${quote(value)})]\n ${rustVariant(value)},`).join("\n")}\n}\n`;
|
||||
const typescript = `// Generated by scripts/generate-acpx-sidecar-contract.mjs. Do not edit.\n\nexport const GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION = ${protocolVersion} as const;\nexport const GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE = ${quote(turnDispatchJournalFile)};\nexport const GENERATED_ACPX_TURN_DISPATCH_SCHEMA = ${quote(turnDispatchSchema)};\nexport const GENERATED_ACPX_SIDECAR_COMMANDS = [\n${commands.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\nexport type GeneratedAcpxSidecarCommand =\n (typeof GENERATED_ACPX_SIDECAR_COMMANDS)[number];\n\nexport const GENERATED_ACPX_SIDECAR_EVENT_TYPES = [\n${events.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\nexport type GeneratedAcpxSidecarEventType =\n (typeof GENERATED_ACPX_SIDECAR_EVENT_TYPES)[number];\n\nexport const GENERATED_ACPX_SIDECAR_OPEN_PARAMS = [\n${openParams.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\n\nexport const GENERATED_ACPX_SIDECAR_PROVIDER_POLICY_FIELDS = [\n${providerPolicyFields.map((value) => ` ${quote(value)},`).join("\n")}\n] as const;\n\nexport type GeneratedAcpxToolOperation =\n | "read"\n | "search"\n | "list"\n | "edit"\n | "execute"\n | "unknown";\n\nexport const GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE = [\n${toolOperationPrecedence.map(([operation, tokens]) => ` { operation: ${quote(operation)}, tokens: [${tokens.map(quote).join(", ")}] },`).join("\n")}\n] as const;\n\nexport function classifyGeneratedAcpxToolOperation(\n toolKind: unknown,\n toolTitle: unknown,\n): GeneratedAcpxToolOperation {\n const candidate =\n typeof toolKind === "string" && toolKind\n ? toolKind\n : typeof toolTitle === "string"\n ? toolTitle\n : "";\n const normalized = candidate.slice(0, 240).toLowerCase();\n for (const { operation, tokens } of GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE) {\n if (tokens.some((token) => normalized.includes(token))) return operation;\n }\n return normalized ? "execute" : "unknown";\n}\n`;
|
||||
const rust = `// Generated by scripts/generate-acpx-sidecar-contract.mjs. Do not edit.\n\npub const GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION: u64 = ${protocolVersion};\npub const GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE: &str = ${quote(turnDispatchJournalFile)};\npub const GENERATED_ACPX_TURN_DISPATCH_SCHEMA: &str = ${quote(turnDispatchSchema)};\n\npub const GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE: &[(&str, &[&str])] = &[\n${toolOperationPrecedence.map(([operation, tokens]) => ` (${quote(operation)}, &[${tokens.map(quote).join(", ")}]),`).join("\n")}\n];\n\npub fn classify_generated_acpx_tool_operation(kind: &str, title: &str) -> &'static str {\n let kind = kind.to_ascii_lowercase();\n let title = title.to_ascii_lowercase();\n let candidate = if kind.is_empty() { &title } else { &kind };\n for (operation, tokens) in GENERATED_ACPX_TOOL_OPERATION_PRECEDENCE {\n if tokens.iter().any(|token| candidate.contains(token)) {\n return operation;\n }\n }\n if candidate.is_empty() {\n "unknown"\n } else {\n "execute"\n }\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub enum GeneratedAcpxSidecarCommand {\n${commands.map((value) => ` ${rustVariant(value)},`).join("\n")}\n}\n\nimpl GeneratedAcpxSidecarCommand {\n pub const fn as_str(self) -> &'static str {\n match self {\n${commands.map((value) => ` Self::${rustVariant(value)} => ${quote(value)},`).join("\n")}\n }\n }\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Deserialize)]\npub enum GeneratedAcpxSidecarEventType {\n${events.map((value) => ` #[serde(rename = ${quote(value)})]\n ${rustVariant(value)},`).join("\n")}\n}\n`;
|
||||
|
||||
// Classification must inspect the same complete provider value in both
|
||||
// languages. Display-field truncation happens only after operation selection.
|
||||
|
||||
@@ -114,6 +114,7 @@ import {
|
||||
} from "./acpx-sidecar-lifecycle.js";
|
||||
|
||||
import { AcpxTurnControlLedger, parseAcpxTurnControl } from "../drivers/acpx/turn-controls.js";
|
||||
import { createAcpxTurnDispatchJournal } from "../drivers/acpx/turn-dispatch-journal.js";
|
||||
|
||||
const turnControls = new AcpxTurnControlLedger();
|
||||
const MAX_PENDING_TOOLS = 512;
|
||||
@@ -488,6 +489,8 @@ async function dispatch(
|
||||
onPermissionRequest: (providerRequest, context) =>
|
||||
waitForPermission(currentTurnId, activeAgent, providerRequest, context, toolEvidence),
|
||||
onTerminalSessionFailure: (failure) => { sessionFailure.value = failure; },
|
||||
// Lets a Runner that lost this request prove whether the prompt left.
|
||||
dispatchJournal: createAcpxTurnDispatchJournal(openParams!.runtimeDirectory, currentTurnId),
|
||||
});
|
||||
} catch (error) {
|
||||
turnId = null;
|
||||
|
||||
@@ -1005,6 +1005,72 @@ describe("Codex ACPX runtime adapter", () => {
|
||||
} finally { await port.close({ reason: "fixture done" }); }
|
||||
});
|
||||
|
||||
it("journals a turn before provider work and its prompt dispatch before the prompt write", async () => {
|
||||
const runtime = fakeRuntime();
|
||||
let created!: AcpRuntimeOptions;
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), { createRegistry: () => registry(), createStore: () => store(),
|
||||
createRuntime: value => { created = value; return runtime; } });
|
||||
const order: string[] = [];
|
||||
const journal = {
|
||||
markReceived: vi.fn(() => { order.push("received"); return true; }),
|
||||
markDispatched: vi.fn(() => { order.push("dispatched"); }),
|
||||
};
|
||||
const pending = pendingExtensionTurn("journaled-turn");
|
||||
vi.mocked(runtime.startTurn).mockImplementation(() => { order.push("runtime.startTurn"); return pending.turn; });
|
||||
try {
|
||||
const turn = port.startTurn({ requestId: "journaled-turn", text: "fixture", dispatchJournal: journal });
|
||||
expect(order).toEqual(["received", "runtime.startTurn"]);
|
||||
// Other outbound traffic never counts as the prompt.
|
||||
created.onAcpMessage?.("outbound", { id: 7, method: "session/set_config_option", params: {} });
|
||||
created.onAcpMessage?.("outbound", { method: "session/cancel", params: {} });
|
||||
expect(journal.markDispatched).not.toHaveBeenCalled();
|
||||
created.onAcpMessage?.("outbound", { id: 8, method: "session/prompt", params: { sessionId: "backend-1" } });
|
||||
// Only the first prompt of the turn is journaled.
|
||||
created.onAcpMessage?.("outbound", { id: 9, method: "session/prompt", params: { sessionId: "backend-1" } });
|
||||
expect(order).toEqual(["received", "runtime.startTurn", "dispatched"]);
|
||||
pending.settle();
|
||||
await turn.result;
|
||||
} finally { await port.close({ reason: "fixture done" }); }
|
||||
});
|
||||
|
||||
it("runs a turn unjournaled when its start cannot be journaled", async () => {
|
||||
const runtime = fakeRuntime();
|
||||
let created!: AcpRuntimeOptions;
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), { createRegistry: () => registry(), createStore: () => store(),
|
||||
createRuntime: value => { created = value; return runtime; } });
|
||||
const pending = pendingExtensionTurn("unjournaled-start");
|
||||
vi.mocked(runtime.startTurn).mockImplementation(() => pending.turn);
|
||||
const markDispatched = vi.fn();
|
||||
try {
|
||||
const turn = port.startTurn({ requestId: "unjournaled-start", text: "fixture",
|
||||
dispatchJournal: { markReceived: () => false, markDispatched } });
|
||||
created.onAcpMessage?.("outbound", { id: 1, method: "session/prompt", params: {} });
|
||||
expect(markDispatched).not.toHaveBeenCalled();
|
||||
pending.settle();
|
||||
await turn.result;
|
||||
} finally { await port.close({ reason: "fixture done" }); }
|
||||
});
|
||||
|
||||
it("aborts the prompt write when its dispatch cannot be journaled", async () => {
|
||||
const runtime = fakeRuntime();
|
||||
let created!: AcpRuntimeOptions;
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), { createRegistry: () => registry(), createStore: () => store(),
|
||||
createRuntime: value => { created = value; return runtime; } });
|
||||
const pending = pendingExtensionTurn("unjournaled-turn");
|
||||
vi.mocked(runtime.startTurn).mockImplementation(() => pending.turn);
|
||||
try {
|
||||
const turn = port.startTurn({ requestId: "unjournaled-turn", text: "fixture", dispatchJournal: {
|
||||
markReceived: () => true,
|
||||
markDispatched: () => { throw new Error("ENOSPC: journal write failed"); },
|
||||
} });
|
||||
// ACPX calls this hook before it writes the bytes; a throw rejects that write.
|
||||
expect(() => created.onAcpMessage?.("outbound", { id: 1, method: "session/prompt", params: {} }))
|
||||
.toThrow("ENOSPC: journal write failed");
|
||||
pending.settle();
|
||||
await turn.result;
|
||||
} finally { await port.close({ reason: "fixture done" }); }
|
||||
});
|
||||
|
||||
it.each(["cancel", "closeStream"] as const)("retains the bound Grok usage response during %s until the prompt settles", async (stop) => {
|
||||
const runtime = fakeRuntime();
|
||||
const durableRecord = { acpxRecordId: "record-1", acpSessionId: "backend-1", agentSessionId: "agent-1", lastRequestId: "previous" };
|
||||
|
||||
@@ -45,6 +45,7 @@ import { ACPX_CAPABILITY_PROFILES } from "./capability-profiles.js";
|
||||
import { createCursorInstructionAdmission } from "./cursor-instructions.js";
|
||||
import { createAntigravityAdmission } from "./antigravity-admission.js";
|
||||
import { createAcpxModeBinding } from "./provider-mode.js";
|
||||
import { isAcpPromptRequest, type AcpxTurnDispatchJournal } from "./turn-dispatch-journal.js";
|
||||
import { assertCopilotPromptPolicy, createCopilotProtocolGuard } from "./copilot-policy.js";
|
||||
|
||||
const VERIFIED_COMMAND_SENTINEL = "paperclip-verified-acpx-command";
|
||||
@@ -122,6 +123,8 @@ interface AcpxRuntimeExtensionBoundary {
|
||||
active: AcpxRuntimeExtensionTurn | null;
|
||||
sessionIds: Set<string>;
|
||||
steering: AcpxRuntimeSteeringCapability | null;
|
||||
/** The started turn whose first session/prompt request is not yet written. */
|
||||
promptDispatch: { turn: AcpxRuntimeExtensionTurn; journal: AcpxTurnDispatchJournal } | null;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -326,7 +329,7 @@ export async function openQualifiedAcpxRuntime(
|
||||
const extensionNotifications = new Set(extensionProfile.extensionNotifications);
|
||||
const promptUsage = options.profile.agent === "grok" ? createGrokPromptUsageCollector() : undefined;
|
||||
const extensionBoundary: AcpxRuntimeExtensionBoundary = {
|
||||
active: null, sessionIds: new Set(), steering: null,
|
||||
active: null, sessionIds: new Set(), steering: null, promptDispatch: null,
|
||||
};
|
||||
const ownsExtensionTurn = (active: AcpxRuntimeExtensionTurn, params: Record<string, unknown>): boolean =>
|
||||
extensionBoundary.active === active && !active.signal.aborted
|
||||
@@ -508,6 +511,13 @@ export async function openQualifiedAcpxRuntime(
|
||||
onSessionNotification: acceptGoalNotification,
|
||||
onAcpMessage: (direction, message) => {
|
||||
const active = extensionBoundary.active;
|
||||
// ACPX calls this synchronously before it writes an outbound message to
|
||||
// the provider stream, so a thrown journal error aborts the prompt write.
|
||||
const dispatch = extensionBoundary.promptDispatch;
|
||||
if (direction === "outbound" && dispatch && dispatch.turn === active && isAcpPromptRequest(message)) {
|
||||
extensionBoundary.promptDispatch = null;
|
||||
dispatch.journal.markDispatched();
|
||||
}
|
||||
if (active && (direction === "inbound" || !active.signal.aborted)) promptUsage?.observe(direction, message);
|
||||
},
|
||||
spawnEnvironment: () => ({
|
||||
@@ -1552,6 +1562,8 @@ function runtimePort(
|
||||
if (providerProcessLost) throw new AcpxProviderProcessLostError();
|
||||
assertPromptPolicy?.(input.text);
|
||||
if (extensionBoundary.active) throw new Error("ACPX runtime already has an active turn");
|
||||
// Record the turn before any provider work can send its prompt.
|
||||
const dispatchJournal = input.dispatchJournal?.markReceived() ? input.dispatchJournal : null;
|
||||
const approval = new AbortController();
|
||||
const controller = new AbortController();
|
||||
const providerDeath = new AbortController();
|
||||
@@ -1563,6 +1575,7 @@ function runtimePort(
|
||||
onPlanProposal: input.onPlanProposal,
|
||||
};
|
||||
extensionBoundary.active = extensionTurn;
|
||||
extensionBoundary.promptDispatch = dispatchJournal ? { turn: extensionTurn, journal: dispatchJournal } : null;
|
||||
promptUsage?.begin(input.requestId, identity.backendSessionId);
|
||||
// A guardian can outlive the provider while ACP awaits a callback.
|
||||
// Native exit or a typed ACP agent disconnect retires this turn; losing
|
||||
@@ -1585,6 +1598,7 @@ function runtimePort(
|
||||
stopWatchingProviderExit();
|
||||
controller.abort(new Error("ACPX extension turn expired"));
|
||||
if (extensionBoundary.active === extensionTurn) extensionBoundary.active = null;
|
||||
if (extensionBoundary.promptDispatch?.turn === extensionTurn) extensionBoundary.promptDispatch = null;
|
||||
};
|
||||
permissionBoundary.hasAdmittedTurn = true;
|
||||
permissionBoundary.active = approval;
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
// Generated by scripts/generate-acpx-sidecar-contract.mjs. Do not edit.
|
||||
|
||||
export const GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION = 2 as const;
|
||||
export const GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE = "paperclip-turn-dispatch.jsonl";
|
||||
export const GENERATED_ACPX_TURN_DISPATCH_SCHEMA = "paperclip.acpx-turn-dispatch.v1";
|
||||
export const GENERATED_ACPX_SIDECAR_COMMANDS = [
|
||||
"initialize",
|
||||
"session.open",
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { retainAcpxCredentialRefresh } from "./credential-refresh.js";
|
||||
import type { AcpxTurnDispatchJournal } from "./turn-dispatch-journal.js";
|
||||
import { assertAcpxExecutionAdmission, parseAcpxCandidateStartupAdmission, type AcpxCandidateStartupAdmission } from "./candidate-admission.js";
|
||||
import { resolveGrokDisableWebSearch } from "./grok-web-search.js";
|
||||
import { recordAcpxAdmissionFailure, type ACPX_ADMISSION_CODES } from "./admission-diagnostics.js";
|
||||
@@ -143,6 +144,8 @@ export interface AcpxRuntimeTurnInput {
|
||||
onPlanProposal?: (proposal: AcpxPlanProposal) => void;
|
||||
/** Receives the typed AIR failure behind ACPX's generic terminal error. */
|
||||
onTerminalSessionFailure?: (failure: AcpxTerminalSessionFailure) => void;
|
||||
/** Records the turn start and the prompt dispatch durably (turn-dispatch-journal.ts). */
|
||||
dispatchJournal?: AcpxTurnDispatchJournal;
|
||||
}
|
||||
|
||||
export interface AcpxRuntimeTurn {
|
||||
@@ -933,6 +936,7 @@ export class AcpxRuntimeHost {
|
||||
...(input.onExtensionNotification ? { onExtensionNotification: input.onExtensionNotification } : {}),
|
||||
...(input.onPlanProposal ? { onPlanProposal: input.onPlanProposal } : {}),
|
||||
...(input.onTerminalSessionFailure ? { onTerminalSessionFailure: input.onTerminalSessionFailure } : {}),
|
||||
...(input.dispatchJournal ? { dispatchJournal: input.dispatchJournal } : {}),
|
||||
});
|
||||
this.#activeTurn = turn;
|
||||
const managed = withAcpxTurnCancellation(turn, reason => this.close({ reason }), ACPX_TURN_CANCELLATION_SHUTDOWN_BOUND_MS);
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { lstat, mkdtemp, readFile, rm, symlink, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { createAcpRuntime, createAgentRegistry, createRuntimeStore } from "acpx/runtime";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE, GENERATED_ACPX_TURN_DISPATCH_SCHEMA } from "./generated-sidecar-contract.js";
|
||||
import { createAcpxTurnDispatchJournal, isAcpPromptRequest } from "./turn-dispatch-journal.js";
|
||||
|
||||
const directories: string[] = [];
|
||||
async function directory(): Promise<string> {
|
||||
const value = await mkdtemp(join(tmpdir(), "acpx-turn-dispatch-"));
|
||||
directories.push(value);
|
||||
return value;
|
||||
}
|
||||
afterEach(async () => {
|
||||
await Promise.all(directories.splice(0).map(value => rm(value, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
const lines = async (dir: string) => (await readFile(join(dir, GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE), "utf8"))
|
||||
.split("\n").filter(Boolean).map(line => JSON.parse(line) as Record<string, unknown>);
|
||||
|
||||
describe("ACPX turn dispatch journal", () => {
|
||||
it("keeps one record set per turn: received truncates, dispatched appends", async () => {
|
||||
const dir = await directory();
|
||||
const first = createAcpxTurnDispatchJournal(dir, "turn-1");
|
||||
expect(first.markReceived()).toBe(true);
|
||||
first.markDispatched();
|
||||
expect(await lines(dir)).toEqual([
|
||||
{ schema: GENERATED_ACPX_TURN_DISPATCH_SCHEMA, turnId: "turn-1", phase: "received" },
|
||||
{ schema: GENERATED_ACPX_TURN_DISPATCH_SCHEMA, turnId: "turn-1", phase: "dispatched" },
|
||||
]);
|
||||
createAcpxTurnDispatchJournal(dir, "turn-2").markReceived();
|
||||
expect(await lines(dir)).toEqual([{ schema: GENERATED_ACPX_TURN_DISPATCH_SCHEMA, turnId: "turn-2", phase: "received" }]);
|
||||
});
|
||||
|
||||
it("never records a dispatch without its turn's received record, and never follows a symlink", async () => {
|
||||
const dir = await directory();
|
||||
expect(() => createAcpxTurnDispatchJournal(dir, "turn-1").markDispatched()).toThrow(/ENOENT/);
|
||||
const target = join(dir, "elsewhere");
|
||||
await writeFile(target, "");
|
||||
await symlink(target, join(dir, GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE));
|
||||
// The turn runs unjournaled, and the planted link is removed, not followed.
|
||||
expect(createAcpxTurnDispatchJournal(dir, "turn-1").markReceived()).toBe(false);
|
||||
expect(await readFile(target, "utf8")).toBe("");
|
||||
await expect(lstat(join(dir, GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("leaves no record when the start record cannot be written", async () => {
|
||||
const missing = join(await directory(), "missing");
|
||||
expect(createAcpxTurnDispatchJournal(missing, "turn-1").markReceived()).toBe(false);
|
||||
});
|
||||
|
||||
it("recognizes only the prompt request", () => {
|
||||
expect(isAcpPromptRequest({ jsonrpc: "2.0", id: 3, method: "session/prompt", params: {} })).toBe(true);
|
||||
expect(isAcpPromptRequest({ jsonrpc: "2.0", method: "session/prompt", params: {} })).toBe(false);
|
||||
expect(isAcpPromptRequest({ jsonrpc: "2.0", id: 3, method: "session/load", params: {} })).toBe(false);
|
||||
expect(isAcpPromptRequest({ jsonrpc: "2.0", id: 3, result: {} })).toBe(false);
|
||||
});
|
||||
|
||||
// The proof depends on ACPX calling onAcpMessage("outbound") before it writes
|
||||
// the prompt to the provider. A real ACPX runtime and a controlled ACP peer
|
||||
// check that order: the peer refuses a prompt it receives before the journal
|
||||
// holds its dispatch record.
|
||||
it("is written before a real ACPX runtime sends the prompt to the provider", async () => {
|
||||
const dir = await directory();
|
||||
const journalPath = join(dir, GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE);
|
||||
const peer = String.raw`
|
||||
const fs = require('node:fs');
|
||||
const send = x => process.stdout.write(JSON.stringify({ jsonrpc: '2.0', ...x }) + '\n');
|
||||
require('node:readline').createInterface({ input: process.stdin }).on('line', line => {
|
||||
const m = JSON.parse(line);
|
||||
if (m.method === 'initialize') send({ id: m.id, result: { protocolVersion: 1, agentCapabilities: { loadSession: true }, authMethods: [] } });
|
||||
else if (m.method === 'session/new') send({ id: m.id, result: { sessionId: 'journal-session' } });
|
||||
else if (m.method === 'session/prompt') {
|
||||
const journal = fs.existsSync(process.env.JOURNAL) ? fs.readFileSync(process.env.JOURNAL, 'utf8') : '';
|
||||
if (journal.includes('"phase":"dispatched"')) send({ id: m.id, result: { stopReason: 'end_turn' } });
|
||||
else send({ id: m.id, error: { code: -32603, message: 'prompt arrived before its dispatch record' } });
|
||||
}
|
||||
});`;
|
||||
const children: ChildProcess[] = [];
|
||||
const journal = createAcpxTurnDispatchJournal(dir, "turn-ordered");
|
||||
let pending: typeof journal | null = null;
|
||||
const runtime = createAcpRuntime({
|
||||
cwd: dir,
|
||||
agentRegistry: createAgentRegistry({ overrides: { claude: "fixture" } }),
|
||||
sessionStore: createRuntimeStore({ stateDir: join(dir, "state") }),
|
||||
permissionMode: "deny-all",
|
||||
onAcpMessage: (direction, value) => {
|
||||
if (direction === "outbound" && pending && isAcpPromptRequest(value)) {
|
||||
const current = pending;
|
||||
pending = null;
|
||||
current.markDispatched();
|
||||
}
|
||||
},
|
||||
spawnAgent: () => {
|
||||
const child = spawn(process.execPath, ["-e", peer], { cwd: dir, env: { ...process.env, JOURNAL: journalPath }, stdio: ["pipe", "pipe", "pipe"] });
|
||||
children.push(child);
|
||||
return child;
|
||||
},
|
||||
});
|
||||
try {
|
||||
const handle = await runtime.ensureSession({ sessionKey: "journal-ordering", agent: "claude", mode: "persistent", cwd: dir });
|
||||
journal.markReceived();
|
||||
pending = journal;
|
||||
const turn = runtime.startTurn({ handle, text: "fixture", mode: "prompt", requestId: "turn-ordered", timeoutMs: 5_000 });
|
||||
const drain = (async () => { for await (const _event of turn.events) { /* drain */ } })();
|
||||
expect(await turn.result).toMatchObject({ status: "completed" });
|
||||
await drain;
|
||||
expect((await lines(dir)).map(line => line.phase)).toEqual(["received", "dispatched"]);
|
||||
} finally {
|
||||
for (const child of children) if (child.exitCode === null) child.kill("SIGKILL");
|
||||
await runtime.close?.({ handle: undefined as never, reason: "fixture done" }).catch(() => undefined);
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
// A dispatch that cannot be journaled must never reach the provider, or a
|
||||
// lone `received` record would later prove a falsehood and duplicate it.
|
||||
it("keeps the prompt from a real ACPX provider when its dispatch record cannot be written", async () => {
|
||||
const dir = await directory();
|
||||
const seen = join(dir, "prompt-seen");
|
||||
const peer = String.raw`
|
||||
const fs = require('node:fs');
|
||||
const send = x => process.stdout.write(JSON.stringify({ jsonrpc: '2.0', ...x }) + '\n');
|
||||
require('node:readline').createInterface({ input: process.stdin }).on('line', line => {
|
||||
const m = JSON.parse(line);
|
||||
if (m.method === 'initialize') send({ id: m.id, result: { protocolVersion: 1, agentCapabilities: { loadSession: true }, authMethods: [] } });
|
||||
else if (m.method === 'session/new') send({ id: m.id, result: { sessionId: 'journal-session' } });
|
||||
else if (m.method === 'session/prompt') { fs.appendFileSync(process.env.SEEN, 'prompt\n'); send({ id: m.id, result: { stopReason: 'end_turn' } }); }
|
||||
});`;
|
||||
const children: ChildProcess[] = [];
|
||||
let armed = false;
|
||||
const runtime = createAcpRuntime({
|
||||
cwd: dir,
|
||||
agentRegistry: createAgentRegistry({ overrides: { claude: "fixture" } }),
|
||||
sessionStore: createRuntimeStore({ stateDir: join(dir, "state") }),
|
||||
permissionMode: "deny-all",
|
||||
onAcpMessage: (direction, value) => {
|
||||
if (armed && direction === "outbound" && isAcpPromptRequest(value)) throw new Error("ENOSPC: dispatch record not written");
|
||||
},
|
||||
spawnAgent: () => {
|
||||
const child = spawn(process.execPath, ["-e", peer], { cwd: dir, env: { ...process.env, SEEN: seen }, stdio: ["pipe", "pipe", "pipe"] });
|
||||
children.push(child);
|
||||
return child;
|
||||
},
|
||||
});
|
||||
try {
|
||||
const handle = await runtime.ensureSession({ sessionKey: "journal-refused", agent: "claude", mode: "persistent", cwd: dir });
|
||||
armed = true;
|
||||
const turn = runtime.startTurn({ handle, text: "fixture", mode: "prompt", requestId: "turn-refused", timeoutMs: 5_000 });
|
||||
const drain = (async () => { for await (const _event of turn.events) { /* drain */ } })().catch(() => undefined);
|
||||
const result = await turn.result.catch((error: unknown) => ({ status: "rejected", error }));
|
||||
await drain;
|
||||
expect(result.status).not.toBe("completed");
|
||||
// Give a stray write time to arrive before asserting it never did.
|
||||
await new Promise(resolve => setTimeout(resolve, 200));
|
||||
await expect(readFile(seen, "utf8")).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally {
|
||||
for (const child of children) if (child.exitCode === null) child.kill("SIGKILL");
|
||||
await runtime.close?.({ handle: undefined as never, reason: "fixture done" }).catch(() => undefined);
|
||||
}
|
||||
}, 30_000);
|
||||
});
|
||||
@@ -0,0 +1,92 @@
|
||||
import { closeSync, constants, fsyncSync, openSync, unlinkSync, writeSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import {
|
||||
GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE,
|
||||
GENERATED_ACPX_TURN_DISPATCH_SCHEMA,
|
||||
} from "./generated-sidecar-contract.js";
|
||||
|
||||
/**
|
||||
* Durable proof of whether a turn's prompt reached the provider.
|
||||
*
|
||||
* A turn start can be cut off (a timeout, a frozen or killed sidecar) after
|
||||
* the Runner asked for it and before it heard back. The Runner then cannot
|
||||
* tell whether the provider received the prompt: resending could duplicate it
|
||||
* and a continuation could lose it. The runtime port records two facts in the
|
||||
* session's private runtime directory:
|
||||
*
|
||||
* - `received`: the port started the turn. Written first, before any provider
|
||||
* work, truncating the previous turn's record. Not synced: losing it only
|
||||
* loses the proof.
|
||||
* - `dispatched`: appended and fsynced synchronously immediately before the
|
||||
* turn's first `session/prompt` request is written to the provider. A failed
|
||||
* write aborts the prompt write.
|
||||
*
|
||||
* The Runner reads the file only after it proves that the sidecar and the
|
||||
* provider are gone, so neither can still write. A lone `received` line then
|
||||
* proves that the prompt was never sent.
|
||||
*/
|
||||
export interface AcpxTurnDispatchJournal {
|
||||
/**
|
||||
* Returns false when the record could not be written. The turn then runs
|
||||
* unjournaled: without a `received` record no reader can form a proof.
|
||||
*/
|
||||
markReceived(): boolean;
|
||||
markDispatched(): void;
|
||||
}
|
||||
|
||||
function record(turnId: string, phase: "received" | "dispatched"): string {
|
||||
return `${JSON.stringify({ schema: GENERATED_ACPX_TURN_DISPATCH_SCHEMA, turnId, phase })}\n`;
|
||||
}
|
||||
|
||||
function writeAll(fd: number, line: string): void {
|
||||
const bytes = Buffer.from(line, "utf8");
|
||||
let offset = 0;
|
||||
while (offset < bytes.length) offset += writeSync(fd, bytes, offset, bytes.length - offset);
|
||||
}
|
||||
|
||||
export function createAcpxTurnDispatchJournal(
|
||||
runtimeDirectory: string,
|
||||
turnId: string,
|
||||
): AcpxTurnDispatchJournal {
|
||||
const path = join(runtimeDirectory, GENERATED_ACPX_TURN_DISPATCH_JOURNAL_FILE);
|
||||
return {
|
||||
markReceived() {
|
||||
try {
|
||||
const fd = openSync(path, constants.O_WRONLY | constants.O_CREAT | constants.O_TRUNC | constants.O_NOFOLLOW, 0o600);
|
||||
try {
|
||||
writeAll(fd, record(turnId, "received"));
|
||||
} finally {
|
||||
closeSync(fd);
|
||||
}
|
||||
return true;
|
||||
} catch (error) {
|
||||
// A journal problem must never block the turn, but a `received`
|
||||
// record left behind without its dispatch would prove a falsehood.
|
||||
// Remove whatever was written; if even that fails, fail the start.
|
||||
try {
|
||||
unlinkSync(path);
|
||||
} catch (cleanup) {
|
||||
if ((cleanup as NodeJS.ErrnoException).code !== "ENOENT") throw error;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
},
|
||||
markDispatched() {
|
||||
// O_APPEND without O_CREAT: a missing `received` line must not be
|
||||
// replaced by a lone `dispatched` record that a reader cannot bind.
|
||||
const fd = openSync(path, constants.O_WRONLY | constants.O_APPEND | constants.O_NOFOLLOW);
|
||||
try {
|
||||
writeAll(fd, record(turnId, "dispatched"));
|
||||
fsyncSync(fd);
|
||||
} finally {
|
||||
closeSync(fd);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** True for the JSON-RPC request that hands a turn's prompt to the provider. */
|
||||
export function isAcpPromptRequest(message: unknown): boolean {
|
||||
return typeof message === "object" && message !== null && "id" in message
|
||||
&& (message as { method?: unknown }).method === "session/prompt";
|
||||
}
|
||||
@@ -294,3 +294,40 @@ it.skipIf(process.platform === "win32")("settles an ACPX read tail through ordin
|
||||
expect(await f.count("processStart")).toBe(3);
|
||||
} finally { await f.cleanup(); }
|
||||
}, 30_000);
|
||||
|
||||
// A turn start cut off in flight (here: the sidecar dies mid-start). The
|
||||
// sidecar's dispatch journal decides what the controller's next turn/start
|
||||
// may do: resend once when the prompt never left, refuse when it may have.
|
||||
it.skipIf(process.platform === "win32").each(["before", "after"] as const)("resends an ACPX turn whose start was cut off %s its prompt left, only when the journal proves it", async stage => {
|
||||
const f = await fixture("claude");
|
||||
try {
|
||||
const source = f.recover({ authorityRunId: f.identity.runId });
|
||||
expect((await source.transport.request("thread/read", {})).thread).toMatchObject(f.thread);
|
||||
await f.settle();
|
||||
await writeFile(f.fault, stage === "before" ? "hang-before-dispatch" : "hang-after-dispatch");
|
||||
const journal = join(f.root, "acpx", "paperclip-turn-dispatch.jsonl");
|
||||
const sidecar = (await f.calls()).filter(entry => entry.processStart).at(-1)!.processStart as number;
|
||||
const cut = source.transport.request("turn/start", { input: [{ type: "text", text: "Write the brief." }] })
|
||||
.then(() => null, (error: unknown) => error);
|
||||
await vi.waitFor(async () => {
|
||||
const phases = (await readFile(journal, "utf8")).trim().split("\n").map(line => JSON.parse(line).phase);
|
||||
expect(phases).toEqual(stage === "before" ? ["received"] : ["received", "dispatched"]);
|
||||
}, { timeout: 10_000 });
|
||||
kill(sidecar);
|
||||
expect(String(await cut)).toContain(stage === "before" ? "before its prompt reached the provider" : "failed closed");
|
||||
await writeFile(f.fault, "");
|
||||
const startsBefore = await f.count("turn.start");
|
||||
if (stage === "before") {
|
||||
await f.complete(source); await f.settle();
|
||||
expect(await f.count("turn.start")).toBe(startsBefore + 1);
|
||||
} else {
|
||||
await expect(source.transport.request("turn/start", { input: [{ type: "text", text: "Write the brief." }] }))
|
||||
.rejects.toThrow("ACPX session is not open");
|
||||
expect(await f.count("turn.start")).toBe(startsBefore);
|
||||
}
|
||||
} catch (error) {
|
||||
const diagnostics = (await readFile(join(f.root, "native.stderr"), "utf8").catch(() => ""))
|
||||
+ (await readFile(join(f.root, "diagnostics", "runnerd.stderr.log"), "utf8").catch(() => ""));
|
||||
throw new Error(`${String(error)}\n${diagnostics}`, { cause: error });
|
||||
} finally { await f.cleanup(); }
|
||||
}, 60_000);
|
||||
Reference in new issue
Block a user