Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-09 07:10:34 +01:00
Code Issues Packages Projects Releases Wiki Activity

Enforce Codex receipt-limit deadlines while Core is offline (#19353)

Browse Source 
## Thinking Path

> - Paperclip uses a Runner to execute agent work.
> - The Runner keeps a record of each tool call, called a receipt.
> - When that storage reaches its limit, the Runner asks Codex to stop.
> - Core can stop responding while the stop request is pending.
> - Codex may already have written its completion behind earlier output.
> - The Runner must keep its stop deadline and preserve that real
completion.

## Linked Issues or Issue Description

Refs #19029. This is its first isolated slice; #19029 remains unmerged.
Related #17594 covers controller replacement. Related #17516 retains
receipts during explicit stop; this change handles the autonomous
receipt-limit deadline.

## What Changed

- Keep the existing 2s/30s stop deadlines while Core is offline,
including when a connection waits for a reply. Other lease and runtime
deadlines still apply. The Runner sleeps between connection attempts and
deadline checks.
- Stop the owned local Codex process when its deadline expires. Read its
remaining output within the existing size and message limits, then
process it in order. Only after output has ended and all earlier
messages have been processed can the Runner report a missing completion.
- After the local process has stopped as requested, skip replies that
only reject or cancel old requests. Commands, user responses and actual
tool results still require successful delivery.
- Keep the stop behavior, regression tests and transport documentation
in this slice.

## Verification

Standalone checks on this branch:

- Seven existing Codex tool-record limit integration cases pass.
- Twenty focused checks cover deadline cleanup and retained output,
including 273 distinct messages followed by one real completion, missing
completion, and a pipe that stays open.
- Separate checks cover strict tool-result delivery, three process-stop
cases, and lease/runtime/reconnect time limits.
- Four controller process tests pass. They include Core accepting a
connection but never replying, and a reconnect delay longer than the
allowed recovery period.
- Existing warm attachment still retains the completed turn's remaining
output and usage.
- Rust formatting, diff checks and the local secret scan pass.

The extraction first omitted an existing test helper and the connected
retry preamble. Standalone compilation and receipt tests caught both
omissions. Restoring them produced the results above. Full regression
checks run in GitHub CI.

Hosted CI passed both native platforms and typecheck, but [server11
failed](https://github.com/paperclipai/paperclip/actions/runs/42118725143/job/127565828671)
one unchanged status-card test. Its fake clock advances while real
database work runs; the final clock assertion was 50 ms past its limit.
The identical test, service and scheduler passed all 36 cases in [the
earlier baseline
shard](https://github.com/paperclipai/paperclip/actions/runs/42111889388/job/127541685625).
The local comparison first needed a built SDK, then stopped at embedded
Postgres startup before any test ran. It did not reproduce the
assertion. The single failed-jobs-only retry passed. The original
failure and the blocked local comparison remain recorded here; the retry
does not establish a precise timing cause.

At exact head `f453509e6f`, all 59 hosted checks completed: 55 succeeded
and four were skipped. Snyk also passed. Standard review round 1 and
selected review round 2 each scored 5/5 on this head, with no unresolved
threads.

## Risks

The Runner now stops its local Codex process at the original deadline
even while Core is unavailable. If another program keeps the output pipe
open, or queued output exceeds the existing limits, the Runner cannot
confirm the outcome. It reports that uncertainty and keeps the records
already written to disk. Tests use real child processes with controlled
provider fixtures. Live provider and deployment qualification remain
separate.

## Model Used

OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning effort, through
Codex. Used source inspection, native subprocess tests and targeted
TypeScript tests.

## 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 <noreply@paperclip.ing>
This commit is contained in:
Nicky LeachandPaperclip authored and GitHub committed 2026-11-05 14:47:17 -08:00
1 parent a6ab6a301c
commit b61a815d3d
9 files changed
+1109 -85
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/paperclip-runner/runner/Cargo.toml
+1 -1
View File
@@ -33,7 +33,7 @@ rustls = { version = "0.23", default-features = false, features = ["ring", "std"
rustls-native-certs = "0.8"
rustls-pemfile = "2.2"
reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls"] }
rustix = { version = "1.1", features = ["fs", "net"] }
rustix = { version = "1.1", features = ["fs", "net", "process"] }
sha1 = "0.10"
tungstenite = { version = "0.28", default-features = false, features = ["handshake", "rustls-tls-native-roots"] }
toml_edit = "0.22"
packages/paperclip-runner/runner/DURABLE_TRANSPORT.md
+7
View File
@@ -41,6 +41,13 @@ source sequence the runner has produced; acknowledged prefixes are removed
atomically from durable state. After disconnect, every remaining event is sent
again with the same identity and source sequence.
If Codex hits its tool-receipt limit during reconnect, runnerd stops its local
process group at the original deadline. It reads the remaining stdout in short
intervals, then replays the frames in order across as many batches as needed.
Only confirmed EOF with no terminal permits a missing-terminal result.
If stdout stays open or the buffer lacks room for one maximum-size frame,
runnerd reports uncertainty and keeps the events already committed to disk.
Executors retain polled events until runnerd acknowledges each event after its
outbox commit. Batches commit one event at a time, so a later oversized event or
capacity failure cannot roll back the accepted prefix or discard the
packages/paperclip-runner/runner/crates/runner-core/examples/recovery-fixture.rs
+19 -2
View File
@@ -3,7 +3,7 @@ use std::fs;
use std::path::PathBuf;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;
use std::time::{Duration, Instant};
use paperclip_runner_core::durable::{
capture_bootstrap_ticket, run_durable_runner, Command, CommandExecution, CommandExecutor,
@@ -23,6 +23,10 @@ struct Script {
receipt_file: PathBuf,
#[serde(default)]
accounting: bool,
#[serde(default)]
provider_stop_ms: Option<u64>,
#[serde(default)]
reconnect_delay_ms: Option<u64>,
}
#[derive(Default)]
@@ -129,6 +133,8 @@ fn workload(script: Script, shared: Shared) {
struct ScriptedExecutor {
shared: Shared,
worker: Option<thread::JoinHandle<()>>,
stop_after: Option<Duration>,
stop_at: Option<Instant>,
}
impl ScriptedExecutor {
@@ -137,6 +143,8 @@ impl ScriptedExecutor {
let worker_shared = shared.clone();
Self {
shared,
stop_after: script.provider_stop_ms.map(Duration::from_millis),
stop_at: None,
worker: Some(thread::spawn(move || workload(script, worker_shared))),
}
}
@@ -151,6 +159,7 @@ impl CommandExecutor for ScriptedExecutor {
return Err(DurableRunnerError::invalid("fixture turn started twice"));
}
work.started = true;
self.stop_at = self.stop_after.map(|delay| Instant::now() + delay);
}
"turn.steer" => work.resumed = true,
"session.snapshot" => {}
@@ -170,6 +179,14 @@ impl CommandExecutor for ScriptedExecutor {
self.retained_events()
}
fn maintain_disconnected_provider(&mut self) -> Result<Option<Instant>, DurableRunnerError> {
if self.stop_at.is_some_and(|due| due <= Instant::now()) {
self.shutdown()?;
self.stop_at = None;
}
Ok(self.stop_at)
}
fn retained_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
Ok(self
.shared
@@ -247,7 +264,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
max_outbox_bytes: value("--max-outbox-bytes")?.parse()?,
p0_reserve_bytes: value("--p0-reserve-bytes")?.parse()?,
max_frame_bytes: 1024 * 1024,
reconnect_delay: Duration::from_millis(20),
reconnect_delay: Duration::from_millis(script.reconnect_delay_ms.unwrap_or(20)),
reconnect_grace: Some(Duration::from_millis(
value("--reconnect-grace-ms")?.parse()?,
)),
packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs
+253 -9
View File
@@ -835,6 +835,13 @@ pub struct CodexProvider {
exit_drain: Option<ProviderExitDrain>,
}
#[derive(Debug)]
pub(crate) struct FencedProviderOutput {
pub stdout_drained: bool,
pub output_empty: bool,
pub next_drain_deadline: Option<std::time::Instant>,
}
struct ProviderExitDrain {
process_generation: u64,
deadline: std::time::Instant,
@@ -2482,7 +2489,7 @@ impl CodexProvider {
} else {
json!({"id": rpc_id, "error": {"code": -32000, "message": message}})
};
self.send_frame(&response)?;
self.send_courtesy_response(&response)?;
if self.notification_identity_diagnostics >= 32 {
return Ok(None);
}
@@ -2521,7 +2528,7 @@ impl CodexProvider {
// The terminal notification is already authoritative and may be
// waiting in the durable outbox. A courtesy rejection must not turn a
// provider that has closed stdin into a fatal polling error.
let _ = self.send_frame(&response);
let _ = self.send_courtesy_response(&response);
Ok(Some(CodexProviderEvent::Notification {
method: "warning".to_owned(),
params: json!({"message": message, "providerMethod": bounded_method(method)}),
@@ -2829,7 +2836,7 @@ impl CodexProvider {
"Codex tool name",
)?;
if !self.authorized_tool_ids.contains(&operation_id) {
self.send_frame(&json!({
self.send_courtesy_response(&json!({
"id": rpc_id,
"result": codex_tool_failure("Paperclip did not authorize this tool for the run"),
}))?;
@@ -3016,7 +3023,7 @@ impl CodexProvider {
if self.pending_runtime_requests.len() >= MAX_PENDING_RUNTIME_REQUESTS
|| retained_request_bytes.is_none()
{
self.send_frame(&json!({
self.send_courtesy_response(&json!({
"id": rpc_id,
"error": {
"code": -32000,
@@ -3059,7 +3066,7 @@ impl CodexProvider {
question_set,
}));
}
self.send_frame(&json!({
self.send_courtesy_response(&json!({
"id": rpc_id,
"error": {"code": -32601, "message": "provider request is unavailable in this runner layer"},
}))?;
@@ -3302,6 +3309,21 @@ impl CodexProvider {
}
pub fn deliver_tool_result(&mut self, result: &ToolResult) -> Result<(), LocalRunnerError> {
self.deliver_tool_result_with_courtesy(result, false)
}
pub(crate) fn deliver_tool_rejection(
&mut self,
result: &ToolResult,
) -> Result<(), LocalRunnerError> {
self.deliver_tool_result_with_courtesy(result, true)
}
fn deliver_tool_result_with_courtesy(
&mut self,
result: &ToolResult,
courtesy_rejection: bool,
) -> Result<(), LocalRunnerError> {
let pending = self
.pending_tool_requests
.get(&result.call_id)
@@ -3324,13 +3346,18 @@ impl CodexProvider {
}
let text = String::from_utf8(result_bytes)
.expect("serde_json always serializes JSON values as valid UTF-8");
self.send_frame(&json!({
let response = json!({
"id": pending.rpc_id,
"result": {
"success": !result.is_error,
"contentItems": [{"type": "inputText", "text": text}],
},
}))?;
});
if courtesy_rejection && result.is_error {
self.send_courtesy_response(&response)?;
} else {
self.send_frame(&response)?;
}
if let Some(completed) = self.pending_tool_requests.remove(&result.call_id) {
self.pending_tool_request_bytes = self
.pending_tool_request_bytes
@@ -3354,6 +3381,92 @@ impl CodexProvider {
}
}
#[cfg(test)]
pub(crate) fn process_exited(&mut self) -> Result<bool, LocalRunnerError> {
self.process.try_wait().map(|exit| exit.is_some())
}
/// Fence only the already-owned local process. Keep every buffered frame
/// for connected interpretation; no courtesy response or provider RPC runs.
/// Reserve one maximum frame before reading; a full FIFO cannot certify EOF.
pub(crate) fn fence_local_receipt_limit(
&mut self,
) -> Result<FencedProviderOutput, LocalRunnerError> {
if self.config.provider != "codex" {
return Err(LocalRunnerError::invalid(
"local receipt-limit fence requires Codex",
));
}
self.expected_shutdown = true;
self.process.terminate_group_immediately()?;
if self
.exit_drain
.as_ref()
.is_none_or(|drain| drain.process_generation != self.process_generation)
{
self.exit_drain = Some(ProviderExitDrain {
process_generation: self.process_generation,
deadline: std::time::Instant::now() + self.process.shutdown_grace(),
timed_out: false,
warning_emitted: false,
});
}
let tick = std::time::Instant::now() + Duration::from_millis(5);
while !self.process.stdout_drained() && std::time::Instant::now() < tick {
if self
.pending_messages
.len()
.saturating_add(self.deferred_ambiguous_messages.len())
>= MAX_BUFFERED_MESSAGES
|| self.pending_message_bytes
> MAX_BUFFERED_MESSAGE_BYTES.saturating_sub(CODEX_APP_SERVER_MAX_FRAME_BYTES)
{
return Err(LocalRunnerError::invalid(
"Fenced Codex output reached its bounded raw buffer; EOF remains unverified",
));
}
let wait = Duration::from_millis(1)
.min(tick.saturating_duration_since(std::time::Instant::now()));
let Some(line) = self.process.receive_stdout_line(wait)? else {
break;
};
let trace_frame_id = self.trace_inbound(&line);
let message = parse_provider_message(&line)?;
let bytes = json_size(&message, "buffered Codex provider message")?;
self.pending_message_bytes =
retain_buffered_message_bytes(self.pending_message_bytes, bytes).ok_or_else(
|| {
LocalRunnerError::invalid(
"Codex buffered messages exceed the 16 MiB aggregate limit",
)
},
)?;
self.pending_messages.push_back(BufferedProviderMessage {
value: message,
trace_frame_id,
});
}
let drain = self
.exit_drain
.as_ref()
.expect("fenced process has a drain deadline");
let stdout_drained = !self.process.stdout_failed() && self.process.stdout_drained();
if drain.timed_out || !stdout_drained && std::time::Instant::now() >= drain.deadline {
return Err(LocalRunnerError::invalid("Fenced Codex stdout did not reach verified EOF within shutdown grace; output requires reconciliation"));
}
Ok(FencedProviderOutput {
stdout_drained,
output_empty: stdout_drained
&& self.pending_messages.is_empty()
&& self.deferred_ambiguous_messages.is_empty(),
next_drain_deadline: (!stdout_drained).then_some(
drain
.deadline
.min(std::time::Instant::now() + Duration::from_millis(25)),
),
})
}
pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> {
self.shutdown_with_cursor_cloud_delivery(|_| Ok(false))
}
@@ -3439,7 +3552,7 @@ impl CodexProvider {
} else {
json!({"answers": {}})
};
if let Err(error) = self.send_frame(&json!({
if let Err(error) = self.send_courtesy_response(&json!({
"id": request.rpc_id,
"result": result,
})) {
@@ -3447,7 +3560,7 @@ impl CodexProvider {
}
}
for request in pending.into_values() {
if let Err(error) = self.send_frame(&json!({
if let Err(error) = self.send_courtesy_response(&json!({
"id": request.rpc_id,
"result": codex_tool_failure("Paperclip stopped the active provider turn"),
})) {
@@ -3465,6 +3578,18 @@ impl CodexProvider {
.map_err(ProviderRequestError::into_inner)
}
fn send_courtesy_response(&mut self, value: &Value) -> Result<(), LocalRunnerError> {
// A reply cannot affect a deliberately stopped local conversation.
// Commands and authoritative results must still use strict delivery.
if self.config.provider == "codex"
&& self.expected_shutdown
&& self.process.try_wait()?.is_some()
{
return Ok(());
}
self.send_frame(value)
}
fn send_frame(&mut self, value: &Value) -> Result<(), LocalRunnerError> {
if let Some(trace) = self.trace.as_mut() {
let raw = serde_json::to_vec(value).unwrap_or_default();
@@ -4796,6 +4921,125 @@ done
provider
}
#[cfg(unix)]
#[test]
fn fenced_tool_rejection_keeps_identity_and_real_delivery_strict() {
let mut provider = completion_tail_provider();
provider.expected_shutdown = false;
provider.pending_tool_requests.insert(
"denied-call".to_owned(),
PendingToolRequest {
rpc_id: json!(91),
operation_id: "denied-operation".to_owned(),
input: json!({}),
retained_bytes: 0,
},
);
let mut rejection = ToolResult {
call_id: "denied-call".to_owned(),
operation_id: "denied-operation".to_owned(),
result: json!({"error": "denied"}),
is_error: true,
};
provider.process.terminate_group_immediately().unwrap();
assert!(provider.deliver_tool_rejection(&rejection).is_err());
provider.fence_local_receipt_limit().unwrap();
assert!(provider.deliver_tool_result(&rejection).is_err());
assert!(provider.request("thread/read", json!({})).is_err());
provider.config.provider = "cursor_cloud".to_owned();
assert!(provider.deliver_tool_rejection(&rejection).is_err());
provider.config.provider = "codex".to_owned();
rejection.operation_id = "foreign-operation".to_owned();
assert!(provider.deliver_tool_rejection(&rejection).is_err());
rejection.operation_id = "denied-operation".to_owned();
rejection.is_error = false;
assert!(provider.deliver_tool_rejection(&rejection).is_err());
rejection.is_error = true;
provider.deliver_tool_rejection(&rejection).unwrap();
assert!(provider.pending_tool_requests.is_empty());
assert!(provider.completed_tool_call_ids.contains("denied-call"));
}
#[cfg(unix)]
#[test]
fn fenced_receipt_output_keeps_prefix_when_stdout_never_closes() {
let mut provider = completion_tail_provider();
let (sender, receiver) = mpsc::channel();
let _original = provider.process.replace_output_receiver_for_test(receiver);
let prefix = json!({"method":"configWarning","params":{"message":"retained prefix"}});
sender
.send(ProcessOutput::Stdout(prefix.to_string()))
.unwrap();
let status = provider.fence_local_receipt_limit().unwrap();
assert!(!status.stdout_drained && !status.output_empty);
assert!(
status.next_drain_deadline.unwrap()
<= std::time::Instant::now() + Duration::from_millis(25)
);
assert!(
status.next_drain_deadline.unwrap() <= provider.exit_drain.as_ref().unwrap().deadline
);
let bytes = provider.pending_message_bytes;
assert_eq!(provider.pending_messages.front().unwrap().value, prefix);
provider.exit_drain.as_mut().unwrap().deadline = std::time::Instant::now();
assert!(provider
.fence_local_receipt_limit()
.unwrap_err()
.to_string()
.contains("EOF"));
assert_eq!(provider.pending_messages.front().unwrap().value, prefix);
assert_eq!(provider.pending_message_bytes, bytes);
}
#[cfg(unix)]
#[test]
fn fenced_receipt_output_reports_capacity_without_dropping_known_frames() {
for (count, text_bytes, eof_waiting) in [
(MAX_BUFFERED_MESSAGES, 1, false),
(MAX_BUFFERED_MESSAGES, 1, true),
(4, 13 * 1024 * 1024 / 4, false),
] {
let mut provider = completion_tail_provider();
let (sender, receiver) = mpsc::channel();
let _original = provider.process.replace_output_receiver_for_test(receiver);
// A full FIFO cannot consume a possibly unstoreable next frame,
// even if this controlled reader has only EOF waiting.
if eof_waiting {
sender.send(ProcessOutput::StdoutClosed).unwrap();
}
for index in 0..count {
let value = json!({"method":"configWarning","params":{"index":index,"message":"x".repeat(text_bytes)}});
provider.pending_message_bytes += json_size(&value, "fixture").unwrap();
provider
.pending_messages
.push_back(BufferedProviderMessage {
value,
trace_frame_id: Some(index as u64),
});
}
let bytes = provider.pending_message_bytes;
assert!(provider
.fence_local_receipt_limit()
.unwrap_err()
.to_string()
.contains("bounded raw buffer"));
assert_eq!(provider.pending_messages.len(), count);
assert_eq!(provider.pending_message_bytes, bytes);
assert_eq!(
provider.pending_messages.front().unwrap().trace_frame_id,
Some(0)
);
assert_eq!(
provider.pending_messages.back().unwrap().trace_frame_id,
Some((count - 1) as u64)
);
assert_eq!(
provider.pending_messages.back().unwrap().value["params"]["index"],
count - 1
);
}
}
#[cfg(unix)]
#[test]
fn warm_attachment_ignores_an_unknown_notification_that_names_no_other_turn() {
packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs
+88 -27
View File
@@ -49,7 +49,7 @@ enum CommandLifecycle {
const TERMINAL_RESULT_ACK_TIMEOUT: Duration = Duration::from_secs(2);
const CUMULATIVE_ACK_PERSIST_INTERVAL: usize = 16;
fn sleep_for_reconnect(base: Duration, max_delay: Duration, attempt: &mut u32) {
fn reconnect_delay(base: Duration, max_delay: Duration, attempt: &mut u32) -> Duration {
let multiplier = 1_u128 << (*attempt).min(5);
let uncapped = base.as_millis().saturating_mul(multiplier);
let capped = uncapped.clamp(1, 5_000) as u64;
@@ -58,23 +58,14 @@ fn sleep_for_reconnect(base: Duration, max_delay: Duration, attempt: &mut u32) {
.map_or(0, |duration| u64::from(duration.subsec_nanos()));
let jitter_percent = 75 + nanos % 51;
*attempt = attempt.saturating_add(1);
let delay = Duration::from_millis(capped.saturating_mul(jitter_percent) / 100).min(max_delay);
if !delay.is_zero() {
thread::sleep(delay);
}
}
fn sleep_before_deadline(delay: Duration, deadline: Instant) {
let bounded_delay = delay.min(deadline.saturating_duration_since(Instant::now()));
if !bounded_delay.is_zero() {
thread::sleep(bounded_delay);
}
Duration::from_millis(capped.saturating_mul(jitter_percent) / 100).min(max_delay)
}
fn connection_attempt_deadline(
config: &DurableRunnerConfig,
started: Instant,
disconnected_since: Option<Instant>,
lease_expires_at_unix_ms: Option<u64>,
) -> Instant {
let now = Instant::now();
// Bound each connection/auth attempt independently of a productive
@@ -91,6 +82,11 @@ fn connection_attempt_deadline(
remaining =
remaining.min(grace.saturating_sub(now.saturating_duration_since(disconnected_at)));
}
if let (Some(expires_at), Ok(now_unix_ms)) = (lease_expires_at_unix_ms, current_unix_ms()) {
remaining = remaining.min(Duration::from_millis(
expires_at.saturating_sub(now_unix_ms),
));
}
now + remaining
}
@@ -333,6 +329,12 @@ pub trait CommandExecutor {
Ok(Vec::new())
}
/// Advance already-owned local cleanup without provider RPCs or ordinary intake.
/// A pending deadline bounds both controller backoff and authentication.
fn maintain_disconnected_provider(&mut self) -> Result<Option<Instant>, DurableRunnerError> {
Ok(None)
}
fn retained_runner_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
Ok(Vec::new())
}
@@ -498,6 +500,7 @@ pub fn run_durable_runner<E: CommandExecutor>(
let mut authenticated_once = false;
let mut disconnected_since: Option<Instant> = None;
let mut reconnect_attempt = 0_u32;
let mut next_connection_attempt = Instant::now();
loop {
if authenticated_once {
@@ -546,8 +549,42 @@ pub fn run_durable_runner<E: CommandExecutor>(
));
}
let provider_deadline = if authenticated_once
&& state.warm_transition.is_none()
&& state.pending_terminal_delivery.is_none()
&& state.pending_provider_cleanup.is_none()
{
executor.maintain_disconnected_provider()?
} else {
None
};
if Instant::now() < next_connection_attempt {
let wake_at = next_connection_attempt.min(connection_attempt_deadline(
&config,
started,
disconnected_since,
lease
.as_ref()
.map(|credential| credential.expires_at_unix_ms),
));
let wake_at = provider_deadline.map_or(wake_at, |due| due.min(wake_at));
let remaining = wake_at.saturating_duration_since(Instant::now());
if !remaining.is_zero() {
thread::sleep(remaining);
}
continue;
}
let using_bootstrap = lease.is_none();
let connect_deadline = connection_attempt_deadline(&config, started, disconnected_since);
let connect_deadline = connection_attempt_deadline(
&config,
started,
disconnected_since,
lease
.as_ref()
.map(|credential| credential.expires_at_unix_ms),
);
let connect_deadline =
provider_deadline.map_or(connect_deadline, |due| due.min(connect_deadline));
let connection = AuthenticatedTransport::connect(
&endpoint,
&config,
@@ -559,7 +596,7 @@ pub fn run_durable_runner<E: CommandExecutor>(
let (mut transport, welcome) = match connection {
Ok(Some(connection)) => connection,
Ok(None) => {
sleep_before_deadline(config.reconnect_delay, connect_deadline);
next_connection_attempt = Instant::now() + config.reconnect_delay;
continue;
}
Err(error) => {
@@ -576,11 +613,12 @@ pub fn run_durable_runner<E: CommandExecutor>(
"bootstrap connection failed closed; provide a fresh one-use ticket",
));
}
sleep_for_reconnect(
config.reconnect_delay,
connect_deadline.saturating_duration_since(Instant::now()),
&mut reconnect_attempt,
);
next_connection_attempt = Instant::now()
+ reconnect_delay(
config.reconnect_delay,
connect_deadline.saturating_duration_since(Instant::now()),
&mut reconnect_attempt,
);
continue;
}
};
@@ -863,9 +901,18 @@ pub fn run_durable_runner<E: CommandExecutor>(
disconnected_since.get_or_insert_with(Instant::now);
state.reconnect_count = state.reconnect_count.saturating_add(1);
store.save(&state)?;
let reconnect_deadline =
connection_attempt_deadline(&config, started, disconnected_since);
sleep_before_deadline(config.reconnect_delay, reconnect_deadline);
let reconnect_deadline = connection_attempt_deadline(
&config,
started,
disconnected_since,
lease
.as_ref()
.map(|credential| credential.expires_at_unix_ms),
);
next_connection_attempt = Instant::now()
+ config
.reconnect_delay
.min(reconnect_deadline.saturating_duration_since(Instant::now()));
continue;
}
loop {
@@ -1167,8 +1214,18 @@ pub fn run_durable_runner<E: CommandExecutor>(
}
}
disconnected_since.get_or_insert_with(Instant::now);
let reconnect_deadline = connection_attempt_deadline(&config, started, disconnected_since);
sleep_before_deadline(config.reconnect_delay, reconnect_deadline);
let reconnect_deadline = connection_attempt_deadline(
&config,
started,
disconnected_since,
lease
.as_ref()
.map(|credential| credential.expires_at_unix_ms),
);
next_connection_attempt = Instant::now()
+ config
.reconnect_delay
.min(reconnect_deadline.saturating_duration_since(Instant::now()));
}
}
@@ -2368,14 +2425,18 @@ mod tests {
config.validate().unwrap();
let started = Instant::now() - Duration::from_secs(21 * 24 * 60 * 60);
let before = Instant::now();
let deadline = connection_attempt_deadline(&config, started, None);
let deadline = connection_attempt_deadline(&config, started, None, None);
assert!(deadline >= before + Duration::from_secs(29));
assert!(deadline <= Instant::now() + Duration::from_secs(30));
config.reconnect_grace = Some(Duration::from_secs(5));
let deadline = connection_attempt_deadline(&config, started, Some(Instant::now()));
let deadline = connection_attempt_deadline(&config, started, Some(Instant::now()), None);
assert!(deadline <= Instant::now() + Duration::from_secs(5));
let lease_expires_at = current_unix_ms().unwrap() + 2_000;
let deadline = connection_attempt_deadline(&config, started, None, Some(lease_expires_at));
assert!(deadline <= Instant::now() + Duration::from_secs(2));
assert!(connection_attempt_deadline(&config, started, None, Some(1)) <= Instant::now());
config.max_runtime = Duration::from_secs(7 * 24 * 60 * 60);
assert!(connection_attempt_deadline(&config, started, None) <= Instant::now());
assert!(connection_attempt_deadline(&config, started, None, None) <= Instant::now());
config.max_runtime = Duration::from_secs(30 * 24 * 60 * 60);
config.validate().unwrap();
}
packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs
+14
View File
@@ -1,4 +1,5 @@
use std::path::PathBuf;
use std::time::Instant;
use serde_json::Value;
@@ -60,6 +61,13 @@ impl CommandExecutor for SelectedExecutor {
}
}
fn maintain_disconnected_provider(&mut self) -> Result<Option<Instant>, DurableRunnerError> {
match self {
Self::LocalFacade(executor) => executor.maintain_disconnected_provider(),
_ => Ok(None),
}
}
fn rotate_authority(&mut self, config: &DurableRunnerConfig) {
match self {
Self::LocalFacade(executor) => executor.rotate_authority(config),
@@ -392,6 +400,12 @@ impl CommandExecutor for NativeProviderCommandExecutor {
.map_or(events, |event| vec![event]))
}
fn maintain_disconnected_provider(&mut self) -> Result<Option<Instant>, DurableRunnerError> {
self.selected
.as_mut()
.map_or(Ok(None), CommandExecutor::maintain_disconnected_provider)
}
fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> {
// A provider that has not yet been selected/restored cannot have
// in-process cleanup to advance. Never launch one merely for ACK debt.
packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs
+148
View File
@@ -1268,6 +1268,77 @@ impl SupervisedProcess {
let _ = self.child.kill();
self.wait()
}
/// Fence this owned process group when an existing deadline leaves no grace.
pub fn terminate_group_immediately(&mut self) -> Result<ProcessExitFact, LocalRunnerError> {
if self.finished {
return self.wait();
}
self.stdin.take();
#[cfg(unix)]
let signal = {
use rustix::process::{kill_process_group, Pid, Signal};
let group = i32::try_from(self.process_group_id)
.ok()
.and_then(Pid::from_raw)
.ok_or_else(|| {
LocalRunnerError::invalid("owned process group identity is invalid")
})?;
match kill_process_group(group, Signal::KILL) {
Ok(()) => Ok(()),
Err(rustix::io::Errno::SRCH) => {
if let Some(exit) = self.try_wait()? {
self.finished = true;
return Ok(exit);
}
Err(LocalRunnerError::invalid(
"owned process group disappeared before leader exit was observed",
))
}
Err(error) => Err(LocalRunnerError::invalid(format!(
"failed to signal owned process group: {error}"
))),
}
};
#[cfg(not(unix))]
let signal = {
if let Some(exit) = self.try_wait()? {
self.finished = true;
return Ok(exit);
}
let signal = self.child.kill().map_err(|error| {
LocalRunnerError::invalid(format!("failed to stop owned process: {error}"))
});
if signal.is_err() {
if let Some(exit) = self.try_wait()? {
self.finished = true;
return Ok(exit);
}
}
signal
};
self.reap_after_immediate_signal(signal)
}
fn reap_after_immediate_signal(
&mut self,
signal: Result<(), LocalRunnerError>,
) -> Result<ProcessExitFact, LocalRunnerError> {
signal?;
let deadline = Instant::now() + Duration::from_millis(100);
loop {
if let Some(exit) = self.try_wait()? {
self.finished = true;
return Ok(exit);
}
if Instant::now() >= deadline {
return Err(LocalRunnerError::invalid(
"owned process termination remains unconfirmed",
));
}
thread::sleep(Duration::from_millis(1));
}
}
}
impl Drop for SupervisedProcess {
@@ -1317,6 +1388,83 @@ mod tests {
use super::*;
use std::time::{SystemTime, UNIX_EPOCH};
#[test]
#[cfg(unix)]
fn immediate_group_fence_does_not_wait_for_shutdown_grace() {
let mut process = SupervisedProcess::spawn(
Path::new("/bin/sh"),
&[
"-c".to_owned(),
"trap '' TERM; printf 'ready\\n'; while :; do read -r line; done".to_owned(),
],
Duration::from_secs(2),
1024,
)
.unwrap();
assert_eq!(
process
.receive_stdout_line(Duration::from_secs(1))
.unwrap()
.as_deref(),
Some("ready")
);
let started = Instant::now();
let exit = process.terminate_group_immediately().unwrap();
assert!(started.elapsed() < Duration::from_secs(1));
assert!(!exit.success);
assert_eq!(
process.terminate_group_immediately().unwrap().exit_code,
exit.exit_code
);
}
#[test]
#[cfg(unix)]
fn immediate_group_fence_accepts_observed_leader_exit_and_missing_group() {
let mut process = SupervisedProcess::spawn(
Path::new("/bin/sh"),
&["-c".to_owned(), "exit 0".to_owned()],
Duration::from_secs(2),
1024,
)
.unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
while process.try_wait().unwrap().is_none() {
assert!(Instant::now() < deadline);
thread::sleep(Duration::from_millis(1));
}
assert!(!process.finished);
assert!(process.terminate_group_immediately().unwrap().success);
assert!(process.finished);
}
#[test]
#[cfg(unix)]
fn failed_immediate_signal_does_not_wait_or_claim_process_exit() {
let mut process = SupervisedProcess::spawn(
Path::new("/bin/sh"),
&[
"-c".to_owned(),
"trap '' TERM; printf 'ready\\n'; while :; do read -r line; done".to_owned(),
],
Duration::from_secs(2),
1024,
)
.unwrap();
assert!(process
.receive_stdout_line(Duration::from_secs(1))
.unwrap()
.is_some());
let started = Instant::now();
assert!(process
.reap_after_immediate_signal(Err(LocalRunnerError::invalid("injected signal failure")))
.is_err());
assert!(started.elapsed() < Duration::from_secs(1));
assert!(!process.finished);
assert!(process.try_wait().unwrap().is_none());
process.terminate_group_immediately().unwrap();
}
#[test]
#[cfg(unix)]
fn stdout_eof_is_sticky_across_consumers_while_stderr_remains_open() {
packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs
+553 -42
View File
@@ -4232,7 +4232,7 @@ impl CodexCommandExecutor {
self.provider
.as_mut()
.expect("provider remains present while rejecting its tool call")
.deliver_tool_result(&rejection)
.deliver_tool_rejection(&rejection)
.map_err(|delivery_error| {
DurableRunnerError::invalid(format!(
"failed to return the semantic tool rejection: {delivery_error}"
@@ -4313,15 +4313,22 @@ impl CodexCommandExecutor {
}
fn settle_receipt_limit_interrupt_after_deadline(&mut self) -> Result<(), DurableRunnerError> {
let interrupt_accepted = self
.state
.as_ref()
.is_some_and(|state| state.receipt_limit_interrupt_accepted);
let provider_shutdown_failed = self
.provider
.as_mut()
.is_some_and(|provider| provider.shutdown().is_err());
self.settle_receipt_limit_after_shutdown(provider_shutdown_failed)
}
fn settle_receipt_limit_after_shutdown(
&mut self,
provider_shutdown_failed: bool,
) -> Result<(), DurableRunnerError> {
self.provider = None;
let interrupt_accepted = self
.state
.as_ref()
.is_some_and(|state| state.receipt_limit_interrupt_accepted);
let identity = self.event_identity()?;
let state = self
.state
@@ -4381,44 +4388,40 @@ impl CodexCommandExecutor {
self.save_state()
}
fn retry_receipt_limit_interrupt(&mut self) -> Result<(), DurableRunnerError> {
let (should_retry, attempts, persisted_deadline) =
self.state.as_ref().map_or((false, 0, None), |state| {
(
state.receipt_limit_interrupt_pending
&& state.active_provider_turn_id.is_some(),
state.receipt_limit_interrupt_attempts,
state.receipt_limit_interrupt_deadline_unix_ms,
)
});
if !should_retry {
return Ok(());
}
let now_unix_ms = current_unix_ms()?;
let deadline_unix_ms = match persisted_deadline {
Some(deadline) => deadline,
None => {
// Older durable state did not record this additive field. Give
// an already-pending interruption one complete bounded window
// after recovery rather than falling back on poll count alone.
let timeout_ms = if self
.state
.as_ref()
.is_some_and(|state| state.receipt_limit_interrupt_accepted)
{
RECEIPT_LIMIT_ACCEPTED_TERMINAL_DEADLINE_MS
} else {
RECEIPT_LIMIT_INTERRUPT_TERMINAL_DEADLINE_MS
};
let deadline = receipt_limit_deadline_after(timeout_ms)?;
self.state
.as_mut()
.expect("Codex state remains available while adding its receipt-limit deadline")
.receipt_limit_interrupt_deadline_unix_ms = Some(deadline);
self.save_state()?;
deadline
}
fn receipt_limit_interrupt_deadline(&mut self) -> Result<Option<u64>, DurableRunnerError> {
let Some(state) = self.state.as_ref().filter(|state| {
state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some()
}) else {
return Ok(None);
};
if let Some(deadline) = state.receipt_limit_interrupt_deadline_unix_ms {
return Ok(Some(deadline));
}
// Preserve the existing legacy-journal window, once, without retrying a provider RPC.
let timeout = if state.receipt_limit_interrupt_accepted {
RECEIPT_LIMIT_ACCEPTED_TERMINAL_DEADLINE_MS
} else {
RECEIPT_LIMIT_INTERRUPT_TERMINAL_DEADLINE_MS
};
let deadline = receipt_limit_deadline_after(timeout)?;
self.state
.as_mut()
.expect("pending interruption has state")
.receipt_limit_interrupt_deadline_unix_ms = Some(deadline);
self.save_state()?;
Ok(Some(deadline))
}
fn retry_receipt_limit_interrupt(&mut self) -> Result<(), DurableRunnerError> {
let Some(deadline_unix_ms) = self.receipt_limit_interrupt_deadline()? else {
return Ok(());
};
let attempts = self
.state
.as_ref()
.expect("pending interruption has state")
.receipt_limit_interrupt_attempts;
let now_unix_ms = current_unix_ms()?;
// Attempts bound network traffic, while the durable wall-clock deadline
// gives an accepted asynchronous interruption time to deliver its
// authoritative terminal. The provider is always polled once more below
@@ -4465,6 +4468,24 @@ impl CodexCommandExecutor {
return Ok(());
};
if current_unix_ms()? >= deadline {
if self
.state
.as_ref()
.is_some_and(|state| state.config.provider == "codex")
{
if let Some(provider) = self.provider.as_mut() {
// A bounded poll may leave a real terminal behind its raw
// prefix. Fence now, but drain it across poll/ACK batches
// before a missing-terminal fallback can replace evidence.
let drained = provider
.fence_local_receipt_limit()
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
if drained.output_empty {
self.settle_receipt_limit_after_shutdown(false)?;
}
return Ok(());
}
}
self.settle_receipt_limit_interrupt_after_deadline()?;
}
Ok(())
@@ -4785,6 +4806,25 @@ impl CodexCommandExecutor {
let receipt_limit_terminal_poll = self.state.as_ref().is_some_and(|state| {
state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some()
});
let local_deadline = self
.state
.as_ref()
.filter(|state| receipt_limit_terminal_poll && state.config.provider == "codex")
.and_then(|state| state.receipt_limit_interrupt_deadline_unix_ms);
if let Some(deadline) = local_deadline {
if current_unix_ms()? >= deadline {
if let Some(provider) = self.provider.as_mut() {
// Verify EOF into bounded raw storage before slow per-event
// persistence can exhaust the ordinary process-exit grace.
let fenced = provider
.fence_local_receipt_limit()
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
if !fenced.stdout_drained {
return Ok(());
}
}
}
}
if !receipt_limit_terminal_poll
&& self
.state
@@ -5478,6 +5518,46 @@ impl CommandExecutor for CodexCommandExecutor {
.collect())
}
fn maintain_disconnected_provider(
&mut self,
) -> Result<Option<std::time::Instant>, DurableRunnerError> {
// Local Codex ownership cannot prove termination of a remote facade.
// Never restore a provider merely to service its cleanup deadline.
if self.provider.is_none()
|| !self.state.as_ref().is_some_and(|state| {
state.config.provider == "codex" && state.startup_attempt.is_none()
})
{
return Ok(None);
}
let Some(deadline) = self.receipt_limit_interrupt_deadline()? else {
return Ok(None);
};
let remaining = deadline.saturating_sub(current_unix_ms()?);
if remaining > 0 {
return std::time::Instant::now()
.checked_add(std::time::Duration::from_millis(remaining))
.map(Some)
.ok_or_else(|| {
DurableRunnerError::invalid(
"Codex receipt-limit deadline is outside the local clock",
)
});
}
let drained = self
.provider
.as_mut()
.expect("local provider remains owned")
.fence_local_receipt_limit()
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
// FIFO evidence may contain a real terminal. Keep it for ordinary
// connected processing instead of manufacturing a conflicting fallback.
if drained.output_empty {
self.settle_receipt_limit_after_shutdown(false)?;
}
Ok(drained.next_drain_deadline)
}
fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> {
if self.state.as_ref().is_some_and(|state| {
state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some()
@@ -7872,6 +7952,437 @@ mod tests {
recovered.validate().unwrap();
}
#[cfg(unix)]
fn disconnected_receipt_limit_fixture(
mode: &str,
deadline: u64,
) -> (CodexCommandExecutor, PathBuf, u64) {
let directory = std::env::temp_dir().join(format!(
"paperclip-offline-receipt-limit-{}",
uuid::Uuid::new_v4()
));
fs::create_dir_all(&directory).unwrap();
let ready = directory.join("ready");
let script = r#"
trap '' TERM
while IFS= read -r line; do
case "$line" in
*'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{}}' ;;
*'"method":"thread/start"'*) printf '%s\n' '{"id":2,"result":{"thread":{"id":"thread-1"}}}' ;;
*'"method":"turn/start"'*)
printf '%s\n' '{"id":3,"result":{"turn":{"id":"turn-1"}}}'
case "$1" in prefix*) printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"predecessor output"}}' ;; esac
case "$1" in many*)
count=0
while [ "$count" -lt "$4" ]; do
printf '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"prefix-%s"}}\n' "$count"
count=$((count + 1))
done ;;
esac
case "$1" in requests_terminal)
printf '%s\n' \
'{"method":"item/completed","params":{"threadId":"thread-1","turnId":"turn-1","item":{"type":"collabAgentToolCall","tool":"spawnAgent","status":"completed","senderThreadId":"thread-1","receiverThreadIds":["child-thread"]}}}' \
'{"id":91,"method":"item/tool/call","params":{"threadId":"child-thread"}}' \
'{"id":92,"method":"item/tool/requestUserInput","params":{"threadId":"child-thread","questions":"malformed"}}' \
'{"id":93,"method":"mcpServer/elicitation/request","params":{"threadId":"child-thread","_meta":{"codex_approval_kind":"mcp_tool_call"}}}' \
'{"id":94,"method":"unsupported/withoutParams"}' \
'{"id":95,"method":"unsupported/malformedParams","params":"malformed"}' \
'{"id":96}' \
'{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"after denied child requests"}}'
;;
esac
case "$1" in *terminal) printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}' ;; esac
printf ready > "$2"
while :; do sleep 1; done ;;
*'"id":'*) printf '%s\n' "$line" >> "$3" ;;
esac
done
"#;
let mut state = opencode_result_state();
state.config.provider = "codex".to_owned();
state.config.driver = "codex_app_server".to_owned();
state.config.command = PathBuf::from("/bin/sh");
state.config.args = vec![
"-c".to_owned(),
script.to_owned(),
"fixture".to_owned(),
mode.to_owned(),
ready.to_string_lossy().into_owned(),
directory
.join("unexpected-rpc")
.to_string_lossy()
.into_owned(),
(MAX_EVENTS_PER_POLL * 2 + 17).to_string(),
];
state.config.model = None;
state.goal_capability = Some(SessionGoalCapability::codex_available());
let mut provider = CodexProvider::start(&state.config, None).unwrap();
provider.start_turn("Test", &state.config.cwd).unwrap();
let until = std::time::Instant::now() + std::time::Duration::from_secs(5);
while !ready.exists() {
assert!(std::time::Instant::now() < until);
std::thread::sleep(std::time::Duration::from_millis(1));
}
state.provider_process_generation = provider.process_generation();
state
.begin_receipt_limit_stop(
"limited-call".to_owned(),
"fixture-tool".to_owned(),
deadline,
)
.unwrap();
state.record_receipt_limit_interrupt_attempt().unwrap();
let mut executor = CodexCommandExecutor::new(&directory);
executor.event_identity = Some(ProviderEventIdentity {
runner_instance_id: "runner".to_owned(),
run_id: "run".to_owned(),
normalized_session_id: "session".to_owned(),
turn_id: "turn".to_owned(),
item_id: "item".to_owned(),
});
executor.state = Some(state);
executor.provider = Some(provider);
executor.save_state().unwrap();
(executor, directory, deadline)
}
#[cfg(unix)]
#[test]
fn disconnected_receipt_limit_preserves_a_future_deadline_without_rpc_or_facade_claims() {
let (mut executor, directory, deadline) =
disconnected_receipt_limit_fixture("silent", current_unix_ms().unwrap() + 60_000);
for _ in 0..2 {
assert!(
executor.maintain_disconnected_provider().unwrap().unwrap()
> std::time::Instant::now()
);
assert_eq!(
executor
.state
.as_ref()
.unwrap()
.receipt_limit_interrupt_deadline_unix_ms,
Some(deadline)
);
assert_eq!(
executor
.state
.as_ref()
.unwrap()
.receipt_limit_interrupt_attempts,
1
);
assert!(!executor
.provider
.as_mut()
.unwrap()
.process_exited()
.unwrap());
}
// A local proxy handle cannot attest that a remote provider was stopped.
executor.state.as_mut().unwrap().config.provider = "cursor_cloud".to_owned();
executor
.state
.as_mut()
.unwrap()
.receipt_limit_interrupt_deadline_unix_ms = Some(1);
assert!(executor.maintain_disconnected_provider().unwrap().is_none());
assert!(
executor
.state
.as_ref()
.unwrap()
.receipt_limit_interrupt_pending
);
assert!(!executor
.provider
.as_mut()
.unwrap()
.process_exited()
.unwrap());
assert!(!directory.join("unexpected-rpc").exists());
assert!(!executor.restore_checked);
executor
.provider
.as_mut()
.unwrap()
.fence_local_receipt_limit()
.unwrap();
fs::remove_dir_all(directory).unwrap();
}
#[cfg(unix)]
#[test]
fn disconnected_receipt_limit_stops_a_hanging_provider_at_its_original_deadline() {
for accepted in [false, true] {
let (mut executor, directory, deadline) =
disconnected_receipt_limit_fixture("silent", 1);
executor
.state
.as_mut()
.unwrap()
.receipt_limit_interrupt_accepted = accepted;
assert_eq!(
executor
.state
.as_ref()
.unwrap()
.receipt_limit_interrupt_deadline_unix_ms,
Some(deadline)
);
let started = std::time::Instant::now();
executor.maintain_disconnected_provider().unwrap();
assert!(started.elapsed() < std::time::Duration::from_secs(1));
if let Some(provider) = executor.provider.as_mut() {
// The owned process is already dead even if the reader has not yet observed EOF.
assert!(provider.process_exited().unwrap());
}
while executor
.state
.as_ref()
.unwrap()
.receipt_limit_interrupt_pending
{
assert!(started.elapsed() < std::time::Duration::from_secs(1));
executor.maintain_disconnected_provider().unwrap();
}
let state = executor.state.as_ref().unwrap();
assert_eq!(state.lifecycle, "closed");
assert!(state.pending_events.iter().any(|event| event.event_type
== if accepted {
"turn.interrupted"
} else {
"turn.failed"
}));
assert!(!directory.join("unexpected-rpc").exists());
assert!(!executor.restore_checked);
fs::remove_dir_all(directory).unwrap();
}
}
#[cfg(unix)]
#[test]
fn disconnected_receipt_limit_preserves_terminal_and_its_fifo_prefix_for_reconnect() {
for mode in ["terminal", "prefix_terminal", "prefix_only"] {
let (mut executor, directory, deadline) = disconnected_receipt_limit_fixture(mode, 1);
let retained = executor.state.as_ref().unwrap().pending_events.clone();
assert!(executor.maintain_disconnected_provider().unwrap().is_none());
assert!(executor
.provider
.as_mut()
.unwrap()
.process_exited()
.unwrap());
let state = executor.state.as_ref().unwrap();
assert_eq!(state.pending_events, retained);
assert!(state.receipt_limit_interrupt_pending);
assert_eq!(state.receipt_limit_interrupt_attempts, 1);
assert_eq!(
state.receipt_limit_interrupt_deadline_unix_ms,
Some(deadline)
);
assert!(!directory.join("unexpected-rpc").exists());
// The ordinary connected path drains the exact prefix before it
// accepts a terminal or applies the honest missing-terminal fallback.
executor.restore_checked = true;
let events = executor.poll_events().unwrap();
let terminal = events
.iter()
.position(|event| {
event.event_type
== if mode == "prefix_only" {
"turn.failed"
} else {
"turn.completed"
}
})
.unwrap();
if mode.starts_with("prefix") {
let prefix = events
.iter()
.position(|event| {
serde_json::to_string(&event.payload)
.unwrap()
.contains("predecessor output")
})
.unwrap();
assert!(prefix < terminal);
}
assert!(
!executor
.state
.as_ref()
.unwrap()
.receipt_limit_interrupt_pending
);
assert!(!directory.join("unexpected-rpc").exists());
fs::remove_dir_all(directory).unwrap();
}
}
#[cfg(unix)]
#[test]
fn fenced_receipt_limit_replays_child_and_unavailable_requests_before_root_terminal() {
let (mut executor, directory, _) =
disconnected_receipt_limit_fixture("requests_terminal", 1);
executor.restore_checked = true;
let mut observed = Vec::new();
for _ in 0..10 {
let events = executor.poll_events().unwrap();
let count = events.len();
observed.extend(events);
executor.acknowledge_events(count).unwrap();
if !executor
.state
.as_ref()
.unwrap()
.receipt_limit_interrupt_pending
{
break;
}
}
let terminal = observed
.iter()
.position(|event| event.event_type == "turn.completed")
.expect("the real root terminal survives every queued courtesy rejection");
assert_eq!(
observed
.iter()
.filter(|event| event.event_type == "turn.completed")
.count(),
1
);
assert!(!observed.iter().any(|event| matches!(
event.event_type.as_str(),
"turn.failed" | "turn.interrupted" | "semantic_tool.input" | "runtime_request.created"
)));
let prefix = observed
.iter()
.position(|event| event.payload["text"] == "after denied child requests")
.unwrap();
assert!(prefix < terminal);
assert!(
observed[..terminal]
.iter()
.filter(|event| {
serde_json::to_string(&event.payload)
.unwrap()
.contains("unsupported Codex request")
|| serde_json::to_string(&event.payload)
.unwrap()
.contains("authorized only for the parent task")
})
.count()
>= 5
);
assert!(!directory.join("unexpected-rpc").exists());
fs::remove_dir_all(directory).unwrap();
}
#[cfg(unix)]
fn assert_disconnected_receipt_limit_drains_batches(mode: &str, has_terminal: bool) {
let (mut executor, directory, _) = disconnected_receipt_limit_fixture(mode, 1);
executor.maintain_disconnected_provider().unwrap();
assert!(executor
.provider
.as_mut()
.unwrap()
.process_exited()
.unwrap());
executor.restore_checked = true;
let mut observed = Vec::new();
let mut batches = 0;
loop {
let events = executor.poll_events().unwrap();
assert!(events.len() <= MAX_EVENTS_PER_POLL + 3);
let count = events.len();
observed.extend(events);
executor.acknowledge_events(count).unwrap();
batches += 1;
assert!(batches <= 10);
let state = executor.state.as_ref().unwrap();
if !state.receipt_limit_interrupt_pending
&& state.pending_events.is_empty()
&& state.queued_events.is_empty()
{
break;
}
}
let texts = observed
.iter()
.filter(|event| event.event_type == "item.delta")
.map(|event| event.payload["text"].as_str().unwrap().to_owned())
.collect::<Vec<_>>();
assert_eq!(
texts.len(),
MAX_EVENTS_PER_POLL * 2 + 17,
"{mode}: {batches} batches, last={:?}",
texts.last()
);
assert!(batches >= 3);
assert_eq!(
texts,
(0..MAX_EVENTS_PER_POLL * 2 + 17)
.map(|index| format!("prefix-{index}"))
.collect::<Vec<_>>()
);
let ids = observed
.iter()
.map(|event| &event.executor_event_id)
.collect::<HashSet<_>>();
assert_eq!(ids.len(), observed.len());
assert_eq!(
observed
.iter()
.filter(|event| event.event_type
== if has_terminal {
"turn.completed"
} else {
"turn.failed"
})
.count(),
1
);
assert!(!observed.iter().any(|event| event.event_type
== if has_terminal {
"turn.failed"
} else {
"turn.completed"
}));
let terminal = observed
.iter()
.position(|event| {
event.event_type
== if has_terminal {
"turn.completed"
} else {
"turn.failed"
}
})
.unwrap();
assert!(
observed
.iter()
.rposition(|event| event.event_type == "item.delta")
.unwrap()
< terminal
);
assert!(!directory.join("unexpected-rpc").exists());
fs::remove_dir_all(directory).unwrap();
}
#[cfg(unix)]
#[test]
fn disconnected_receipt_limit_drains_large_prefix_before_real_terminal() {
assert_disconnected_receipt_limit_drains_batches("many_terminal", true);
}
#[cfg(unix)]
#[test]
fn disconnected_receipt_limit_drains_large_prefix_before_missing_terminal_fallback() {
assert_disconnected_receipt_limit_drains_batches("many_only", false);
}
#[test]
fn receipt_limit_deadline_settlement_preserves_unacknowledged_events() {
let directory = std::env::temp_dir().join(format!(
packages/paperclip-runner/src/control-plane/runner-recovery-fixture.test.ts
+26 -4
View File
@@ -1,6 +1,7 @@
import { execFileSync, spawn, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { createServer, type Socket } from "node:net";
import { tmpdir } from "node:os";
import { resolve } from "node:path";
import { fileURLToPath } from "node:url";
@@ -23,13 +24,13 @@ beforeAll(() => {
"-p", "paperclip-runner-core", "--example", "recovery-fixture"], { timeout: 120_000, stdio: "pipe" });
}, 120_000);
async function fixture(externalPersistence: boolean, graceMs = 20_000) {
async function fixture(externalPersistence: boolean, graceMs = 20_000, providerStopMs?: number, reconnectDelayMs?: number) {
const root = mkdtempSync(resolve(tmpdir(), "runner-in-process-recovery-"));
const stateDirectory = resolve(root, "control-plane");
const receiptFile = resolve(root, "work.json");
const releaseFile = resolve(root, "release");
const scriptFile = resolve(root, "script.json");
writeFileSync(scriptFile, JSON.stringify({ steps: 64, waitAfter: 32, retentionLimit: 8, receiptFile, releaseFile }));
writeFileSync(scriptFile, JSON.stringify({ steps: 64, waitAfter: 32, retentionLimit: 8, receiptFile, releaseFile, providerStopMs, reconnectDelayMs }));
const delivered: Array<{ sourceEventId: string; text: string }> = [];
let checkpoint: DurablePrpCheckpoint | undefined;
const create = () => {
@@ -66,7 +67,7 @@ async function fixture(externalPersistence: boolean, graceMs = 20_000) {
await vi.waitFor(() => expect(core.getCommand(commandId)?.status, `${commandId}: exit=${child?.exitCode}; ${diagnostics}`).toBe("completed"), { timeout: 10_000 });
}
return {
root, receipt, child, exit, delivered, command,
root, port, receipt, child, exit, delivered, command,
checkpoint: () => checkpoint,
release: () => writeFileSync(releaseFile, "compute"),
stopController: async () => {
@@ -127,12 +128,14 @@ it.each([false, true])("preserves useful in-process work through repeated replac
}, 35_000);
it("stops after the original reconnect deadline while retaining recovery state", async () => {
const run = await fixture(true, 1500);
const run = await fixture(true, 1500, undefined, 5000);
try {
await run.command("turn.start", "start-once");
const disconnectedAt = performance.now();
await run.stopController();
run.release();
await run.exit;
expect(performance.now() - disconnectedAt).toBeLessThan(4000);
const state = JSON.parse(readFileSync(resolve(run.root, "runner/runner-state.json"), "utf8"));
expect(state.recoverableFailure).toBe("transport_reconnect_grace_exceeded");
expect(run.receipt()).toMatchObject({ phase: "stopped", pid: run.child.pid });
@@ -140,3 +143,22 @@ it("stops after the original reconnect deadline while retaining recovery state",
expect(run.receipt().peakRetainedEvents).toBeLessThanOrEqual(8);
} finally { await run.cleanup(); }
}, 15_000);
it("enforces local provider cleanup while a replacement controller holds the handshake open", async () => {
const run = await fixture(false, 20_000, 1500);
const sockets = new Set<Socket>();
const silent = createServer(socket => { sockets.add(socket); socket.once("close", () => sockets.delete(socket)); });
try {
await run.command("turn.start", "start-with-deadline");
await run.stopController();
await new Promise<void>((ready, reject) => { silent.once("error", reject); silent.listen(run.port, "127.0.0.1", ready); });
await vi.waitFor(() => expect(sockets.size).toBeGreaterThan(0), { timeout: 4_000 });
await vi.waitFor(() => expect(run.receipt().phase).toBe("stopped"), { timeout: 4_000 });
expect(run.child.exitCode).toBeNull();
} finally {
for (const socket of sockets) socket.destroy();
await new Promise<void>(done => silent.close(() => done()));
await run.cleanup();
}
}, 15_000);