fix(runner): keep healthy native sessions alive (#13261)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native sessions keep a provider process and its work alive across control-plane operations. > - A hidden 15-minute turn deadline stopped work even when the agent timeout was zero. > - A one-hour runner lifetime and fixed connection lease added two more limits. > - Recovery also rejected goal commands because it reconstructed their startup summary with the wrong protocol version. > - This pull request removes implicit duration limits and renews authenticated leases in the harness. > - Healthy sessions can continue without model action or a user interface change. ## Linked Issues or Issue Description Refs #13092 and #12845. Related: #13163 covers sandbox recovery after app restarts; this change covers session duration and lease renewal. **What happened?** A native Codex session stopped after 15 minutes while a tool was still running. The agent had `timeoutSec: 0`. Recovery then rejected a `session.goal.get` startup command with `invalid provider startup ownership fence`. **Expected behavior** An unlimited session keeps working while its provider and authenticated controller remain healthy. Lease maintenance is transparent. Explicit timeouts, cancellation and revoked authority still take effect. **Steps to reproduce** Start a native session with `timeoutSec: 0` and run a tool beyond 15 minutes. Before this fix, the runtime cancels the turn. A recovery startup that uses a goal command also exposes the protocol-version mismatch. ## What Changed - Honor the agent turn timeout. Zero disables the timer. Long explicit durations use timer chunks to avoid Node timer overflow. - Default native runner lifetime to unlimited. Keep bounded startup, reconnect and control-operation deadlines. - Renew leases over the authenticated connection. Persist renewal before the reply. Validate identity, epoch and expiry. Handle duplicate requests and a lost reply on reconnect. - Freeze renewal during warm ownership transitions and terminal handling. - Validate persisted goal startup commands with protocol v2. - Add duration, renewal, ownership, recovery and real-process regression tests. Update runner protocol and recovery docs. - Add no UI components or controls. Renewal requires no model output or user action. ## Verification Current head: `348e369c35c5da8bb8be378f4b35dcf6f40882e7`. [CI run](https://github.com/paperclipai/paperclip/actions/runs/34649767113). - All 32 checks pass on this head. The two Storybook checks are skipped as expected. CI includes full build, typecheck, runner verification, browser suites, server tests and the canary package dry run. - Greptile reports 5/5 on this head. All review threads are resolved, and the security scan passes. - Passed `pnpm -r typecheck` and `pnpm build` locally. - Passed 219 native-runtime and controller tests, including fake-clock tests for three weeks of renewal and 30-day explicit timeouts. Six denial tests confirm that renewal cannot extend expired, revoked or mismatched authority. - Passed 337 executor, cancellation and restart-recovery tests, plus 278 Rust runner-core library tests. - Passed a real runner with a silent fake Codex provider across its original lease expiry. Runner PID, provider PID, thread and active turn stayed unchanged. Warm-attach recovery tests also pass. - Passed all 83 plugin-worker tests and 159 of 161 workspace-runtime tests locally. The two remaining assertions passed with a canonical macOS temporary directory, as did the changed runtime fixture. The full affected server shard passes in CI. - An unchanged GitHub callback-ordering test failed once in CI, passed locally in isolation, and passed its one test-shard retry. The final CI summary is successful. - The full local `pnpm test:run` sweep was interrupted after dependency setup failures and load-related timeouts. Identified failing suites passed in isolated reruns after the dependency repair. The complete test matrix passed remotely in CI. ## Risks - Deploy the controller and runner together to enable renewal. Older peers keep their existing bounded lease behavior. - Unlimited runtime permits long resource use until completion, explicit cancellation, configured timeout or loss of valid authority. - Lease renewal changes authenticated protocol handling. Regression tests cover stale, revoked and mismatched authority, lost replies and warm handoff behavior. - Simulated multi-week tests and a real lease-boundary test do not constitute a weeks-long production soak. ## Model Used OpenAI GPT-6 through Codex, with repository inspection, code execution and TypeScript/Rust test tools. The exact backend revision and context-window size are not exposed in this session. ## 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 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:
Dotta
GitHub
parent
2083bf6f9a
commit
250deab910
|
|
@ -12,6 +12,51 @@ WebSocket acceptance header are not authentication. A production bridge should
|
|||
still use `wss://` for defense in depth and remains a separately reviewed
|
||||
deployment phase.
|
||||
|
||||
## Execution duration and operation deadlines
|
||||
|
||||
Native turns have no total elapsed-time limit unless the operator configures
|
||||
`timeoutSec` on the agent. Zero means unlimited. The controller passes that
|
||||
setting separately from bootstrap, recovery, checkpoint, and finalization
|
||||
operation bounds. A live turn must not inherit the internal 15-minute operation
|
||||
deadline; tool waits count toward an explicitly configured turn duration.
|
||||
|
||||
The native-session API exposes `turnTimeoutMs` for that duration. An explicit
|
||||
`timeoutMs` retains its prior operation/turn behavior for existing embedding
|
||||
callers, while `turnTimeoutMs: 0` overrides its turn bound. Cancellation, budget
|
||||
controls, input handoffs, and bounded cleanup remain independent.
|
||||
|
||||
Normal runner launches also set `--max-runtime-ms 0`, meaning no total process
|
||||
lifetime limit. The standalone durable runner defaults to the same value.
|
||||
Explicit positive limits still stop at their deadline; connection/auth attempts,
|
||||
reconnect grace, cancellation, and idle cleanup remain bounded independently.
|
||||
Neither quiet tool execution nor a productive turn is an idle session.
|
||||
|
||||
The authenticated welcome advertises `connectionLeaseRenewalVersion: 1`.
|
||||
Supporting runners send `lease_renew` halfway through the remaining lease,
|
||||
independently of provider output. The request carries the current expiry and
|
||||
revocation epoch under the exact connection, lease, and run identity. The
|
||||
controller persists an extended expiry before returning `lease_renewed`, which
|
||||
also echoes the request's previous expiry. The runner updates its in-memory
|
||||
lease without replacing its process, provider, thread, turn, token, or epoch.
|
||||
Retries of the same observed expiry replay the persisted extension. Renewal
|
||||
never admits an expired, revoked, or differently bound credential.
|
||||
|
||||
If a reply is lost, reconnect authentication may reconcile a later expiry only
|
||||
when this runner has an outstanding renewal on that same credential. Warm
|
||||
handoff receipts continue to bind exact expiry and renewal pauses during their
|
||||
transition. Old controllers that do not advertise renewal retain their bounded
|
||||
lease behavior; deploying both updated controller and runner is required.
|
||||
|
||||
Tests simulate three weeks of renewal on one authenticated connection, exercise
|
||||
lost-reply reconnect without reexecuting provider startup, and keep a quiet
|
||||
active Codex fixture in the same runner/provider PIDs beyond its original lease
|
||||
expiry. These are boundary regressions, not a weeks-long real-provider soak.
|
||||
|
||||
Provider startup ownership summaries validate goal commands with their required
|
||||
PRP v2 command vocabulary. Incoming wire commands still undergo the negotiated
|
||||
protocol validation before execution; ordinary persisted v1 summaries remain
|
||||
compatible.
|
||||
|
||||
## Connection and authentication
|
||||
|
||||
The connection starts in this order:
|
||||
|
|
|
|||
|
|
@ -193,6 +193,13 @@ These envelopes are local Local runner implementation contracts.
|
|||
- A one-use bootstrap bearer capability returns a short-lived connection lease
|
||||
in `welcome`. Later connections use that lease. Neither raw capability is
|
||||
durable state.
|
||||
- `welcome.payload.connectionLeaseRenewalVersion: 1` opts into authenticated
|
||||
`lease_renew` / `lease_renewed` control frames. Renewal extends the persisted
|
||||
expiry on the same live authority without restarting provider work. Identity,
|
||||
protocol, and revocation epoch remain fixed; expired or revoked leases cannot
|
||||
renew. See [durable recovery](durable-recovery.md#execution-duration-and-operation-deadlines)
|
||||
for retry and warm-handoff rules. Peers lacking this capability retain their
|
||||
original lease expiry.
|
||||
- `hello.resume` reports the last processed controller sequence, next source
|
||||
sequence, cumulative ACK cursor, and current unacknowledged range.
|
||||
- `welcome` selects the one overlapping protocol version, returns the core's
|
||||
|
|
|
|||
|
|
@ -330,7 +330,7 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> {
|
|||
max_frame_bytes: usize_value(args, "--max-frame-bytes", 1024 * 1024)?,
|
||||
reconnect_delay: duration("--reconnect-delay-ms", 250)?,
|
||||
reconnect_grace: optional_u64(args, "--reconnect-grace-ms")?.map(Duration::from_millis),
|
||||
max_runtime: duration("--max-runtime-ms", 60 * 60 * 1000)?,
|
||||
max_runtime: duration("--max-runtime-ms", 0)?,
|
||||
};
|
||||
let executor = NativeProviderCommandExecutor::with_runner_config(state_dir, &config);
|
||||
run_durable_runner(config, ticket, executor)
|
||||
|
|
|
|||
|
|
@ -194,6 +194,7 @@ pub struct DurableRunnerConfig {
|
|||
pub max_frame_bytes: usize,
|
||||
pub reconnect_delay: Duration,
|
||||
pub reconnect_grace: Option<Duration>,
|
||||
/// Zero disables the total process lifetime limit.
|
||||
pub max_runtime: Duration,
|
||||
}
|
||||
|
||||
|
|
@ -245,11 +246,6 @@ impl DurableRunnerConfig {
|
|||
"transport frame limit must be between 1 KiB and 16 MiB",
|
||||
));
|
||||
}
|
||||
if self.max_runtime.is_zero() {
|
||||
return Err(DurableRunnerError::invalid(
|
||||
"durable runner max runtime must be non-zero",
|
||||
));
|
||||
}
|
||||
if self.reconnect_delay.is_zero() || self.reconnect_delay > Duration::from_secs(60) {
|
||||
return Err(DurableRunnerError::invalid(
|
||||
"reconnect delay must be between one millisecond and 60 seconds",
|
||||
|
|
@ -260,11 +256,6 @@ impl DurableRunnerConfig {
|
|||
"reconnect grace must be non-zero when configured",
|
||||
));
|
||||
}
|
||||
if self.max_runtime > Duration::from_secs(7 * 24 * 60 * 60) {
|
||||
return Err(DurableRunnerError::invalid(
|
||||
"durable runner max runtime must not exceed seven days",
|
||||
));
|
||||
}
|
||||
if let Some(profile) = self.acpx_launch_profile.as_ref() {
|
||||
if profile.authority_digest.len() != 71
|
||||
|| !profile.authority_digest.starts_with("sha256:")
|
||||
|
|
|
|||
|
|
@ -77,18 +77,20 @@ fn connection_attempt_deadline(
|
|||
disconnected_since: Option<Instant>,
|
||||
) -> Instant {
|
||||
let now = Instant::now();
|
||||
let runtime_remaining = config
|
||||
.max_runtime
|
||||
.saturating_sub(now.saturating_duration_since(started));
|
||||
let remaining = disconnected_since.zip(config.reconnect_grace).map_or(
|
||||
runtime_remaining,
|
||||
|(disconnected_at, grace)| {
|
||||
runtime_remaining
|
||||
.min(grace.saturating_sub(now.saturating_duration_since(disconnected_at)))
|
||||
},
|
||||
);
|
||||
// Validation caps max_runtime at seven days, and reconnect grace can only
|
||||
// shorten this budget, so adding it to a current Instant cannot overflow.
|
||||
// Bound each connection/auth attempt independently of a productive
|
||||
// session's lifetime. Zero means there is no total runtime deadline.
|
||||
let mut remaining = Duration::from_secs(30);
|
||||
if !config.max_runtime.is_zero() {
|
||||
remaining = remaining.min(
|
||||
config
|
||||
.max_runtime
|
||||
.saturating_sub(now.saturating_duration_since(started)),
|
||||
);
|
||||
}
|
||||
if let Some((disconnected_at, grace)) = disconnected_since.zip(config.reconnect_grace) {
|
||||
remaining =
|
||||
remaining.min(grace.saturating_sub(now.saturating_duration_since(disconnected_at)));
|
||||
}
|
||||
now + remaining
|
||||
}
|
||||
|
||||
|
|
@ -417,7 +419,7 @@ pub fn run_durable_runner<E: CommandExecutor>(
|
|||
));
|
||||
}
|
||||
}
|
||||
if started.elapsed() >= config.max_runtime {
|
||||
if !config.max_runtime.is_zero() && started.elapsed() >= config.max_runtime {
|
||||
let _ = shutdown_preserving_cleanup(&state, &mut executor);
|
||||
record_recoverable_transport_failure(
|
||||
&mut state,
|
||||
|
|
@ -514,7 +516,16 @@ pub fn run_durable_runner<E: CommandExecutor>(
|
|||
state.restore_v2_replay_events(&config)?;
|
||||
}
|
||||
state.last_connection_protocol_version = Some(protocol_version);
|
||||
let connection = welcome.connection;
|
||||
let mut connection = welcome.connection;
|
||||
// Reconnect may follow a durably committed renewal whose reply was
|
||||
// lost. Authentication admits an increased expiry only while our
|
||||
// matching renewal is outstanding; identity and epoch stay exact.
|
||||
if let Some(credential) = lease.as_mut() {
|
||||
credential.expires_at_unix_ms = connection.expires_at_unix_ms;
|
||||
credential.renewal_requested = false;
|
||||
}
|
||||
let mut next_lease_renewal =
|
||||
lease_renewal_deadline(current_unix_ms()?, connection.expires_at_unix_ms);
|
||||
if let Some(transition) = state.warm_transition.clone() {
|
||||
if welcome.warm_transition_version != Some(1) {
|
||||
return Err(DurableRunnerError::invalid(
|
||||
|
|
@ -739,7 +750,7 @@ pub fn run_durable_runner<E: CommandExecutor>(
|
|||
continue;
|
||||
}
|
||||
loop {
|
||||
if started.elapsed() >= config.max_runtime {
|
||||
if !config.max_runtime.is_zero() && started.elapsed() >= config.max_runtime {
|
||||
break;
|
||||
}
|
||||
if let Err(error) = send_outbox(
|
||||
|
|
@ -766,6 +777,38 @@ pub fn run_durable_runner<E: CommandExecutor>(
|
|||
"active connection lease expired; durable state is preserved",
|
||||
));
|
||||
}
|
||||
let now = current_unix_ms()?;
|
||||
if welcome.lease_renewal_version == Some(1) && now >= next_lease_renewal {
|
||||
// A failed write may still have reached the controller.
|
||||
if let Some(credential) = lease.as_mut() {
|
||||
credential.renewal_requested = true;
|
||||
}
|
||||
if let Err(error) = transport.send_json(&control_envelope(
|
||||
&state,
|
||||
&connection,
|
||||
"lease_renew",
|
||||
json!({
|
||||
"connectionLeaseExpiresAtUnixMs": connection.expires_at_unix_ms,
|
||||
"connectionLeaseRevocationEpoch": connection.revocation_epoch,
|
||||
}),
|
||||
)) {
|
||||
disconnected_since.get_or_insert_with(Instant::now);
|
||||
state.record_diagnostic(format!("lease renewal reconnect scheduled: {error}"));
|
||||
state.reconnect_count = state.reconnect_count.saturating_add(1);
|
||||
store.save(&state)?;
|
||||
break;
|
||||
}
|
||||
// Retry a lost reply before expiry without flooding the channel.
|
||||
next_lease_renewal = now.saturating_add(
|
||||
5_000.min(
|
||||
connection
|
||||
.expires_at_unix_ms
|
||||
.saturating_sub(now)
|
||||
.saturating_div(2)
|
||||
.max(1),
|
||||
),
|
||||
);
|
||||
}
|
||||
// Read control before starting another fsynced provider batch.
|
||||
// Consuming the last cumulative ACK must not let a new output
|
||||
// suffix overtake the stop/suspend already queued behind it.
|
||||
|
|
@ -799,6 +842,14 @@ pub fn run_durable_runner<E: CommandExecutor>(
|
|||
break;
|
||||
}
|
||||
match message.get("kind").and_then(Value::as_str) {
|
||||
Some("lease_renewed") => {
|
||||
let credential = lease.as_mut().ok_or_else(|| {
|
||||
DurableRunnerError::invalid("lease renewal requires a live credential")
|
||||
})?;
|
||||
apply_lease_renewal(&message, &mut connection, credential)?;
|
||||
next_lease_renewal =
|
||||
lease_renewal_deadline(current_unix_ms()?, connection.expires_at_unix_ms);
|
||||
}
|
||||
Some("ack") => {
|
||||
let acked = message
|
||||
.pointer("/payload/ackedSourceSeq")
|
||||
|
|
@ -985,6 +1036,47 @@ pub fn run_durable_runner<E: CommandExecutor>(
|
|||
}
|
||||
}
|
||||
|
||||
fn lease_renewal_deadline(now: u64, expires_at: u64) -> u64 {
|
||||
now.saturating_add(expires_at.saturating_sub(now) / 2)
|
||||
}
|
||||
|
||||
fn apply_lease_renewal(
|
||||
message: &Value,
|
||||
connection: &mut ConnectionMetadata,
|
||||
credential: &mut LeaseCredential,
|
||||
) -> Result<(), DurableRunnerError> {
|
||||
let previous = message
|
||||
.pointer("/payload/previousExpiresAtUnixMs")
|
||||
.and_then(Value::as_u64);
|
||||
let expiry = message
|
||||
.pointer("/payload/connectionLeaseExpiresAtUnixMs")
|
||||
.and_then(Value::as_u64);
|
||||
let epoch = message
|
||||
.pointer("/payload/connectionLeaseRevocationEpoch")
|
||||
.and_then(Value::as_u64);
|
||||
let Some(expiry) = expiry else {
|
||||
return Err(DurableRunnerError::invalid(
|
||||
"lease renewal expiry is required",
|
||||
));
|
||||
};
|
||||
if epoch != Some(connection.revocation_epoch)
|
||||
|| expiry < connection.expires_at_unix_ms
|
||||
|| previous.is_none_or(|old| old > connection.expires_at_unix_ms)
|
||||
|| (expiry > connection.expires_at_unix_ms
|
||||
&& (!credential.renewal_requested || previous != Some(connection.expires_at_unix_ms)))
|
||||
|| credential.lease_id != connection.lease_id
|
||||
|| credential.revocation_epoch != connection.revocation_epoch
|
||||
{
|
||||
return Err(DurableRunnerError::invalid(
|
||||
"lease renewal changed the authenticated binding",
|
||||
));
|
||||
}
|
||||
connection.expires_at_unix_ms = expiry;
|
||||
credential.expires_at_unix_ms = expiry;
|
||||
credential.renewal_requested = false;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn persist_lifecycle_before_shutdown<E: CommandExecutor>(
|
||||
state: &mut DurableState,
|
||||
store: &DurableStateStore,
|
||||
|
|
@ -1904,6 +1996,25 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unlimited_lifetime_keeps_attempts_bounded_after_weeks_of_work() {
|
||||
let mut config = config(std::env::temp_dir());
|
||||
config.max_runtime = Duration::ZERO;
|
||||
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);
|
||||
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()));
|
||||
assert!(deadline <= Instant::now() + Duration::from_secs(5));
|
||||
config.max_runtime = Duration::from_secs(7 * 24 * 60 * 60);
|
||||
assert!(connection_attempt_deadline(&config, started, None) <= Instant::now());
|
||||
config.max_runtime = Duration::from_secs(30 * 24 * 60 * 60);
|
||||
config.validate().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn startup_failure_facts_cross_full_fifo_before_failed_command_and_never_poll() {
|
||||
for mode in [
|
||||
|
|
|
|||
|
|
@ -710,6 +710,7 @@ pub(crate) struct LeaseCredential {
|
|||
pub(crate) expires_at_unix_ms: u64,
|
||||
pub(crate) revocation_epoch: u64,
|
||||
token: Secret,
|
||||
pub(crate) renewal_requested: bool,
|
||||
}
|
||||
|
||||
impl LeaseCredential {
|
||||
|
|
@ -734,6 +735,7 @@ pub(crate) struct Welcome {
|
|||
pub(crate) acked_source_seq: Option<u64>,
|
||||
pub(crate) pending_commands: Vec<Command>,
|
||||
pub(crate) warm_transition_version: Option<u64>,
|
||||
pub(crate) lease_renewal_version: Option<u64>,
|
||||
pub(crate) warm_transition: Option<Value>,
|
||||
pub(crate) warm_transition_phase: Option<String>,
|
||||
}
|
||||
|
|
@ -1259,7 +1261,10 @@ fn validate_challenge(
|
|||
match expected_lease {
|
||||
Some(lease)
|
||||
if challenge.credential_lease_id.as_deref() == Some(lease.lease_id.as_str())
|
||||
&& challenge.credential_expires_at_unix_ms == lease.expires_at_unix_ms
|
||||
&& (challenge.credential_expires_at_unix_ms == lease.expires_at_unix_ms
|
||||
|| (lease.renewal_requested
|
||||
&& state.warm_transition.is_none()
|
||||
&& challenge.credential_expires_at_unix_ms > lease.expires_at_unix_ms))
|
||||
&& challenge.revocation_epoch == lease.revocation_epoch => {}
|
||||
None if challenge.credential_lease_id.is_none() => {}
|
||||
_ => {
|
||||
|
|
@ -1369,7 +1374,10 @@ fn validate_welcome(
|
|||
.ok_or_else(|| DurableRunnerError::invalid("welcome revocation epoch is required"))?;
|
||||
if let Some(expected) = expected_lease {
|
||||
if connection_lease_id != expected.lease_id
|
||||
|| expires_at_unix_ms != expected.expires_at_unix_ms
|
||||
|| (expires_at_unix_ms != expected.expires_at_unix_ms
|
||||
&& !(expected.renewal_requested
|
||||
&& state.warm_transition.is_none()
|
||||
&& expires_at_unix_ms > expected.expires_at_unix_ms))
|
||||
|| revocation_epoch != expected.revocation_epoch
|
||||
{
|
||||
return Err(DurableRunnerError::invalid(
|
||||
|
|
@ -1385,6 +1393,7 @@ fn validate_welcome(
|
|||
expires_at_unix_ms,
|
||||
revocation_epoch,
|
||||
token: Secret::new(token),
|
||||
renewal_requested: false,
|
||||
})
|
||||
}
|
||||
None | Some(Value::Null) if credential_kind == "lease" => None,
|
||||
|
|
@ -1422,6 +1431,9 @@ fn validate_welcome(
|
|||
acked_source_seq: payload.get("ackedSourceSeq").and_then(Value::as_u64),
|
||||
pending_commands,
|
||||
warm_transition_version: payload.get("warmTransitionVersion").and_then(Value::as_u64),
|
||||
lease_renewal_version: payload
|
||||
.get("connectionLeaseRenewalVersion")
|
||||
.and_then(Value::as_u64),
|
||||
warm_transition: payload.get("warmTransition").cloned(),
|
||||
warm_transition_phase: payload
|
||||
.get("warmTransitionPhase")
|
||||
|
|
@ -2411,6 +2423,15 @@ mod tests {
|
|||
|
||||
#[test]
|
||||
fn reconnect_replays_unacked_events_and_not_command_effects() {
|
||||
reconnect_without_reexecuting(false);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lost_renewal_reply_reconnects_without_restarting_the_provider() {
|
||||
reconnect_without_reexecuting(true);
|
||||
}
|
||||
|
||||
fn reconnect_without_reexecuting(renewal_reply_lost: bool) {
|
||||
struct EventExecutor {
|
||||
session_open_calls: Arc<AtomicUsize>,
|
||||
shutdown_calls: Arc<AtomicUsize>,
|
||||
|
|
@ -2448,13 +2469,14 @@ mod tests {
|
|||
let mut config = config(port);
|
||||
config.max_runtime = Duration::from_secs(5);
|
||||
let directory = std::env::temp_dir().join(format!(
|
||||
"paperclip-runner-reconnect-fault-{}",
|
||||
"paperclip-runner-reconnect-fault-{}-{renewal_reply_lost}",
|
||||
std::process::id()
|
||||
));
|
||||
let _ = std::fs::remove_dir_all(&directory);
|
||||
config.state_dir = directory.clone();
|
||||
let state = test_state(&config);
|
||||
let expires = current_unix_ms().unwrap() + 60_000;
|
||||
let mut expires =
|
||||
current_unix_ms().unwrap() + if renewal_reply_lost { 2_000 } else { 60_000 };
|
||||
let open_command = json!({
|
||||
"schema": "paperclip.prp.command.v1",
|
||||
"commandId": "command_open",
|
||||
|
|
@ -2489,23 +2511,32 @@ mod tests {
|
|||
revocation_epoch: 0,
|
||||
},
|
||||
);
|
||||
send_secure(
|
||||
&mut first,
|
||||
&mut first_secure,
|
||||
&server_config,
|
||||
&welcome(
|
||||
&server_state,
|
||||
"connection_1",
|
||||
Some("lease-secret"),
|
||||
expires,
|
||||
0,
|
||||
vec![server_open.clone()],
|
||||
),
|
||||
let mut greeting = welcome(
|
||||
&server_state,
|
||||
"connection_1",
|
||||
Some("lease-secret"),
|
||||
expires,
|
||||
0,
|
||||
vec![server_open.clone()],
|
||||
);
|
||||
if renewal_reply_lost {
|
||||
greeting["payload"]["connectionLeaseRenewalVersion"] = json!(1);
|
||||
}
|
||||
send_secure(&mut first, &mut first_secure, &server_config, &greeting);
|
||||
let first_result = receive_secure(&mut first, &mut first_secure, &server_config);
|
||||
let first_event = receive_secure(&mut first, &mut first_secure, &server_config);
|
||||
assert_eq!(first_result["kind"], "command_result");
|
||||
assert_eq!(first_event["kind"], "event");
|
||||
if renewal_reply_lost {
|
||||
let renewal = receive_secure(&mut first, &mut first_secure, &server_config);
|
||||
assert_eq!(renewal["kind"], "lease_renew");
|
||||
assert_eq!(
|
||||
renewal["payload"]["connectionLeaseExpiresAtUnixMs"],
|
||||
json!(expires)
|
||||
);
|
||||
// Commit a new expiry but lose the reply before the runner sees it.
|
||||
expires = current_unix_ms().unwrap() + 60_000;
|
||||
}
|
||||
drop(first);
|
||||
|
||||
let (second_stream, _) = listener.accept().unwrap();
|
||||
|
|
|
|||
|
|
@ -143,8 +143,16 @@ impl ProviderStartupAttempt {
|
|||
!value.is_empty() && value.len() <= limit && !value.chars().any(char::is_control)
|
||||
};
|
||||
let command_valid = self.command.as_ref().is_none_or(|command| {
|
||||
// This is a summary of an already-validated command, not a new
|
||||
// wire request. Goal commands require v2; reconstructing every
|
||||
// summary as v1 rejects a legitimate provider restart on goal/get.
|
||||
let schema = if command.command_type.starts_with("session.goal.") {
|
||||
"paperclip.prp.command.v2"
|
||||
} else {
|
||||
"paperclip.prp.command.v1"
|
||||
};
|
||||
Command {
|
||||
schema: "paperclip.prp.command.v1".to_owned(),
|
||||
schema: schema.to_owned(),
|
||||
command_id: command.command_id.clone(),
|
||||
controller_seq: command.controller_seq,
|
||||
command_type: command.command_type.clone(),
|
||||
|
|
@ -4553,6 +4561,38 @@ mod tests {
|
|||
fs::remove_dir_all(directory).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn goal_command_can_persist_provider_restart_ownership() {
|
||||
for command_type in ["session.goal.get", "session.goal.set", "session.goal.clear"] {
|
||||
let directory = std::env::temp_dir()
|
||||
.join(format!("paperclip-goal-startup-{}", uuid::Uuid::new_v4()));
|
||||
let mut executor = CodexCommandExecutor::new(&directory);
|
||||
executor.state = Some(opencode_result_state());
|
||||
executor.startup_command = Some(ProviderStartupCommand {
|
||||
command_id: "command-goal-recovery".to_owned(),
|
||||
controller_seq: 12,
|
||||
command_type: command_type.to_owned(),
|
||||
});
|
||||
executor
|
||||
.begin_startup(ProviderStartupTrigger::Ensure, 4)
|
||||
.unwrap();
|
||||
executor
|
||||
.observe_startup(ProviderStartupObservation::Spawned {
|
||||
process_id: 123,
|
||||
process_group_id: 123,
|
||||
})
|
||||
.unwrap();
|
||||
let saved: CodexProviderState =
|
||||
serde_json::from_slice(&fs::read(executor.state_path()).unwrap()).unwrap();
|
||||
saved.validate().unwrap();
|
||||
assert_eq!(
|
||||
saved.startup_attempt.unwrap().command.unwrap().command_type,
|
||||
command_type
|
||||
);
|
||||
fs::remove_dir_all(directory).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn startup_phase_fields_are_closed_and_coherent() {
|
||||
let mut attempt = ProviderStartupAttempt {
|
||||
|
|
|
|||
|
|
@ -42,6 +42,100 @@ const identity: DurableRecoveryIdentity = {
|
|||
const expectedRunnerVersion = "0.3.0";
|
||||
const expectedRunnerDigest = `sha256:${"a".repeat(64)}`;
|
||||
|
||||
function renewalRequest(client: AuthenticatedClient, expiresAt: number): Record<string, unknown> {
|
||||
return {
|
||||
protocol: "paperclip.runner", version: client.welcome.version,
|
||||
kind: "lease_renew", ...identity,
|
||||
connectionId: client.welcome.connectionId,
|
||||
connectionLeaseId: client.welcome.connectionLeaseId,
|
||||
payload: {
|
||||
connectionLeaseExpiresAtUnixMs: expiresAt,
|
||||
connectionLeaseRevocationEpoch: (client.welcome.payload as Record<string, unknown>).connectionLeaseRevocationEpoch,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
it("renews one authenticated connection for three weeks without replacing its authority", async () => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "runner-lease-renewal-"));
|
||||
let now = Date.now();
|
||||
const clock = vi.spyOn(Date, "now").mockImplementation(() => now);
|
||||
const ttl = 6 * 60 * 60 * 1_000;
|
||||
const core = new DurablePrpControlPlane({
|
||||
stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest,
|
||||
connectionLeaseTtlMs: ttl,
|
||||
});
|
||||
try {
|
||||
await core.start();
|
||||
const client = (await authenticate(core, core.issueBootstrapTicket()))!;
|
||||
let expiry = Number((client.welcome.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
const leaseId = client.welcome.connectionLeaseId;
|
||||
for (let hour = 0; hour < 21 * 24; hour += 3) {
|
||||
now += ttl / 2;
|
||||
const request = renewalRequest(client, expiry);
|
||||
sendSecure(client, request);
|
||||
const reply = (await receiveSecure(client))!;
|
||||
expect(reply.kind).toBe("lease_renewed");
|
||||
expect(reply.connectionLeaseId).toBe(leaseId);
|
||||
expect(reply.connectionId).toBe(client.welcome.connectionId);
|
||||
const next = Number((reply.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
expect(next).toBe(now + ttl);
|
||||
// A lost reply can be retried without another authority extension.
|
||||
now += 1;
|
||||
sendSecure(client, request);
|
||||
expect((await receiveSecure(client))!.payload).toEqual(reply.payload);
|
||||
expiry = next;
|
||||
}
|
||||
expect(core.store.state.connectionCount).toBe(1);
|
||||
expect(Object.keys(core.store.state.leases)).toHaveLength(1);
|
||||
expect(core.store.state.commands).toEqual([]);
|
||||
client.socket.destroy();
|
||||
await core.stop();
|
||||
const restored = new DurablePrpControlPlane({
|
||||
stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest,
|
||||
connectionLeaseTtlMs: ttl,
|
||||
});
|
||||
try {
|
||||
await restored.start();
|
||||
const resumed = (await authenticate(restored, client.leaseToken!))!;
|
||||
expect(resumed.welcome.connectionLeaseId).toBe(leaseId);
|
||||
expect((resumed.welcome.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs).toBe(expiry);
|
||||
resumed.socket.destroy();
|
||||
} finally { await restored.stop(); }
|
||||
} finally {
|
||||
clock.mockRestore();
|
||||
await core.stop();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["expired", "revoked", "wrong-run", "wrong-connection", "wrong-epoch", "future-expiry"])(
|
||||
"cannot renew a lease with %s authority",
|
||||
async (fault) => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "runner-lease-denial-"));
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest });
|
||||
let clock: ReturnType<typeof vi.spyOn> | undefined;
|
||||
try {
|
||||
await core.start();
|
||||
const client = (await authenticate(core, core.issueBootstrapTicket()))!;
|
||||
const expiry = Number((client.welcome.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
const request = renewalRequest(client, expiry);
|
||||
if (fault === "expired") clock = vi.spyOn(Date, "now").mockReturnValue(expiry);
|
||||
if (fault === "revoked") Object.values(core.store.state.leases)[0]!.revokedAt = new Date().toISOString();
|
||||
if (fault === "wrong-run") request.runId = "different-run";
|
||||
if (fault === "wrong-connection") request.connectionId = "different-connection";
|
||||
if (fault === "wrong-epoch") (request.payload as Record<string, unknown>).connectionLeaseRevocationEpoch = 999;
|
||||
if (fault === "future-expiry") (request.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs = expiry + 1;
|
||||
sendSecure(client, request);
|
||||
expect(await receiveSecure(client)).toBeNull();
|
||||
expect(Object.values(core.store.state.leases)[0]!.expiresAtUnixMs).toBe(expiry);
|
||||
} finally {
|
||||
clock?.mockRestore();
|
||||
await core.stop();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("persists the initial warm attachment seed idempotently and rejects replacement", () => {
|
||||
const root = mkdtempSync(
|
||||
resolve(tmpdir(), "runner-initial-attachment-seed-test-"),
|
||||
|
|
|
|||
|
|
@ -2057,6 +2057,9 @@ export class DurablePrpControlPlane {
|
|||
await this.#authResponse(connection, envelope);
|
||||
return;
|
||||
}
|
||||
// Admit every post-handshake frame against persisted authority before
|
||||
// dispatch, including lease_renew. Renewal cannot revive a revoked or
|
||||
// expired credential or bypass changes to its persisted binding.
|
||||
if (
|
||||
connection.secureChannel === null ||
|
||||
connection.lease === null ||
|
||||
|
|
@ -2068,6 +2071,10 @@ export class DurablePrpControlPlane {
|
|||
connection.close();
|
||||
return;
|
||||
}
|
||||
if (kind === "lease_renew") {
|
||||
this.#renewLease(connection, envelope);
|
||||
return;
|
||||
}
|
||||
if (kind === "event") {
|
||||
if (this.#store.state.warmTransition?.phase === "awaiting_result")
|
||||
connection.replayOnly = true;
|
||||
|
|
@ -2143,6 +2150,62 @@ export class DurablePrpControlPlane {
|
|||
}
|
||||
}
|
||||
|
||||
#renewLease(
|
||||
connection: AuthorityConnection,
|
||||
envelope: Record<string, unknown>,
|
||||
): void {
|
||||
const lease = connection.lease!;
|
||||
const payload = envelope.payload as Record<string, unknown> | undefined;
|
||||
const expectedExpiry = payload?.connectionLeaseExpiresAtUnixMs;
|
||||
if (
|
||||
Object.entries(connection.identity!).some(
|
||||
([key, value]) => envelope[key] !== value,
|
||||
) ||
|
||||
envelope.connectionId !== connection.connectionId ||
|
||||
envelope.connectionLeaseId !== lease.leaseId ||
|
||||
payload?.connectionLeaseRevocationEpoch !== lease.revocationEpoch ||
|
||||
!Number.isSafeInteger(expectedExpiry) ||
|
||||
(expectedExpiry as number) <= 0 ||
|
||||
(expectedExpiry as number) > lease.expiresAtUnixMs
|
||||
) {
|
||||
connection.close();
|
||||
return;
|
||||
}
|
||||
// A handoff receipt binds the exact expiry. Finish that boundary before
|
||||
// renewing; terminal commands likewise retain their existing authority.
|
||||
if (
|
||||
connection.replayOnly ||
|
||||
this.#store.state.warmTransition ||
|
||||
connection.terminalLifecycleCommandId !== null
|
||||
) return;
|
||||
// Repeating a request after a lost reply replays the persisted expiry.
|
||||
// It never extends a credential twice for the same observed generation.
|
||||
if (expectedExpiry === lease.expiresAtUnixMs) {
|
||||
const candidate = structuredClone(this.#store.state);
|
||||
const renewed = candidate.leases[lease.credentialId]!;
|
||||
renewed.expiresAtUnixMs = Math.max(
|
||||
lease.expiresAtUnixMs,
|
||||
Date.now() + this.#connectionLeaseTtlMs,
|
||||
);
|
||||
renewed.expiresAt = new Date(renewed.expiresAtUnixMs).toISOString();
|
||||
candidate.lastLeaseExpiresAt = renewed.expiresAt;
|
||||
this.#store.commit(candidate);
|
||||
connection.lease = this.#store.state.leases[lease.credentialId]!;
|
||||
}
|
||||
connection.sendJson(
|
||||
this.#controlEnvelope(
|
||||
connection,
|
||||
`lease_renewed_${expectedExpiry}`,
|
||||
"lease_renewed",
|
||||
{
|
||||
previousExpiresAtUnixMs: expectedExpiry,
|
||||
connectionLeaseExpiresAtUnixMs: connection.lease!.expiresAtUnixMs,
|
||||
connectionLeaseRevocationEpoch: connection.lease!.revocationEpoch,
|
||||
},
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
#authorizeHello(
|
||||
payload: Record<string, unknown>,
|
||||
): PendingAuthorization | null {
|
||||
|
|
@ -2652,6 +2715,7 @@ export class DurablePrpControlPlane {
|
|||
payload: {
|
||||
selectedVersion: lease.protocolVersion,
|
||||
heartbeatIntervalMs: 250,
|
||||
connectionLeaseRenewalVersion: 1,
|
||||
connectionLeaseId: lease.leaseId,
|
||||
...(leaseToken === null ? {} : { connectionLeaseToken: leaseToken }),
|
||||
connectionLeaseExpiresAt: lease.expiresAt,
|
||||
|
|
|
|||
|
|
@ -3980,6 +3980,58 @@ it("expands coalesced canonical items without dropping strict bindings", () => {
|
|||
]);
|
||||
});
|
||||
|
||||
it("keeps a quiet active Codex turn in the same process across connection lease expiry", async () => {
|
||||
const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-renew-active-"));
|
||||
const callsPath = join(stateDirectory, "calls.log");
|
||||
const cores: DurablePrpControlPlane[] = [];
|
||||
const OriginalCore = durableControlPlane.DurablePrpControlPlane;
|
||||
const coreSpy = vi.spyOn(durableControlPlane, "DurablePrpControlPlane")
|
||||
.mockImplementation(function(options: ConstructorParameters<typeof OriginalCore>[0]) {
|
||||
const core = new OriginalCore({ ...options, connectionLeaseTtlMs: 60_000 });
|
||||
cores.push(core);
|
||||
return core;
|
||||
} as unknown as typeof OriginalCore);
|
||||
const bundle = createCapabilityRunnerdCodexTransport({
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex,
|
||||
codexArgs: fakeCodexArgs(stateDirectory, "--hold-turn", "--record-process-start", "--call-log", callsPath),
|
||||
stateDirectory, runnerReconnectGraceMs: 5_000,
|
||||
});
|
||||
try {
|
||||
const opened = await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] });
|
||||
await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Keep working quietly." }] });
|
||||
const runnerPid = bundle.evidence().runnerPid;
|
||||
const codexPid = bundle.evidence().codexPid;
|
||||
expect(runnerPid).toBeGreaterThan(0);
|
||||
expect(codexPid).toBeGreaterThan(0);
|
||||
const core = cores[0]!;
|
||||
const original = structuredClone(Object.values(core.store.state.leases)[0]!);
|
||||
await vi.waitFor(() => {
|
||||
expect(Date.now()).toBeGreaterThan(original.expiresAtUnixMs + 1_000);
|
||||
}, { timeout: 75_000, interval: 1_000 });
|
||||
const current = Object.values(core.store.state.leases)[0]!;
|
||||
expect(current.expiresAtUnixMs).toBeGreaterThan(original.expiresAtUnixMs);
|
||||
expect(current.leaseId).toBe(original.leaseId);
|
||||
expect(core.store.state.connectionCount).toBe(1);
|
||||
expect(bundle.evidence().runnerPid).toBe(runnerPid);
|
||||
expect(bundle.evidence().codexPid).toBe(codexPid);
|
||||
process.kill(runnerPid!, 0);
|
||||
process.kill(codexPid!, 0);
|
||||
const calls = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/);
|
||||
expect(calls.filter(call => call === "process-start")).toHaveLength(1);
|
||||
expect(calls.filter(call => call === "thread/start")).toHaveLength(1);
|
||||
expect(calls.filter(call => call === "turn/start")).toHaveLength(1);
|
||||
expect(calls).not.toContain("turn/interrupt");
|
||||
expect(calls).not.toContain("thread/resume");
|
||||
const state = JSON.parse(await readFile(join(stateDirectory, "fake-codex-state.json"), "utf8"));
|
||||
expect(state.threadId).toBe(opened.thread.id);
|
||||
expect(state.activeTurnId).toBe("provider-turn-1");
|
||||
} finally {
|
||||
await bundle.transport.close();
|
||||
coreSpy.mockRestore();
|
||||
await rm(stateDirectory, { recursive: true, force: true });
|
||||
}
|
||||
}, 90_000);
|
||||
|
||||
it("runs the lab provider boundary through authenticated durable PRP", async () => {
|
||||
const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-lab-provider-"));
|
||||
const bundle = createCapabilityRunnerdCodexTransport({
|
||||
|
|
|
|||
|
|
@ -4621,7 +4621,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
ticket: core.issueBootstrapTicket(RUNNER_BOOTSTRAP_TICKET_TTL_MS),
|
||||
maxOutboxBytes: RUNNERD_MAX_OUTBOX_BYTES,
|
||||
p0ReserveBytes: RUNNERD_P0_RESERVE_BYTES,
|
||||
maxRuntimeMs: 60 * 60 * 1_000,
|
||||
maxRuntimeMs: 0,
|
||||
reconnectGraceMs: this.options.runnerReconnectGraceMs,
|
||||
lifecyclePolicy: this.options.lifecyclePolicy,
|
||||
runnerBinaryPath,
|
||||
|
|
@ -5244,7 +5244,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
ticket: bootstrapTicket!,
|
||||
maxOutboxBytes: RUNNERD_MAX_OUTBOX_BYTES,
|
||||
p0ReserveBytes: RUNNERD_P0_RESERVE_BYTES,
|
||||
maxRuntimeMs: 60 * 60 * 1_000,
|
||||
maxRuntimeMs: 0,
|
||||
reconnectGraceMs: this.options.runnerReconnectGraceMs,
|
||||
lifecyclePolicy: this.options.lifecyclePolicy,
|
||||
runnerBinaryPath,
|
||||
|
|
|
|||
|
|
@ -201,6 +201,88 @@ function highestContiguous(events: PrpEvent[]): number {
|
|||
}
|
||||
|
||||
describe("executeNativeSession recovery", () => {
|
||||
it.each([undefined, 0, 7 * 24 * 60 * 60 * 1000, 30 * 24 * 60 * 60 * 1000])(
|
||||
"honors long-lived turn duration independently of operation bounds (%s)",
|
||||
async (turnTimeoutMs) => {
|
||||
vi.useFakeTimers();
|
||||
let finish = () => {};
|
||||
const waiting = new Promise<void>((resolve) => { finish = resolve; });
|
||||
const capabilities = {
|
||||
resume: true, typedEvents: true, steering: false,
|
||||
interruption: true, structuredResult: true,
|
||||
};
|
||||
const cancel = vi.fn(async () => { finish(); });
|
||||
const close = vi.fn(async () => { finish(); });
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
async capabilities() { return capabilities; },
|
||||
async *events() {
|
||||
yield runnerEvent(1, "tool.execution.started", { name: "wait for CI", status: "running" });
|
||||
await waiting;
|
||||
yield runnerEvent(2, "turn.completed");
|
||||
},
|
||||
async startTurn() { return { turnId: "turn-recovery" }; },
|
||||
async result() { return { result, terminal, turnId: "turn-recovery" }; },
|
||||
async snapshot() {
|
||||
return { backendKind: "mock", sessionId: "driver-recovery", identity,
|
||||
providerSessionId: "provider-recovery", cursor: null, activeTurnId: null,
|
||||
pendingRuntimeRequests: [], lineage: [] };
|
||||
},
|
||||
cancel, close,
|
||||
};
|
||||
const appendEvent = vi.fn<ControlPlanePort["appendEvent"]>(async event => ({
|
||||
cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq,
|
||||
disposition: "committed",
|
||||
}));
|
||||
const completeRun = vi.fn(async () => {});
|
||||
try {
|
||||
const execution = executeNativeSession({
|
||||
input, turnTimeoutMs,
|
||||
backend: {
|
||||
async descriptor() { return { kind: "mock", name: "long-lived", version: "1", capabilities }; },
|
||||
async openSession() { return session; },
|
||||
},
|
||||
controlPlane: {
|
||||
async openRun() {}, async checkpointSession() {}, appendEvent,
|
||||
async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; },
|
||||
completeRun,
|
||||
},
|
||||
runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery",
|
||||
});
|
||||
// Observe rejection before advancing the clock, including on the old implementation.
|
||||
let failure: unknown;
|
||||
const observed = execution.catch(error => { failure = error; return null; });
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(appendEvent).toHaveBeenCalledOnce();
|
||||
await vi.advanceTimersByTimeAsync(6 * 24 * 60 * 60 * 1000);
|
||||
expect(failure).toBeUndefined();
|
||||
expect(cancel).not.toHaveBeenCalled();
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(completeRun).not.toHaveBeenCalled();
|
||||
if (turnTimeoutMs) {
|
||||
// Includes a 30-day bound beyond Node's single-timer maximum.
|
||||
await vi.advanceTimersByTimeAsync(turnTimeoutMs - 6 * 24 * 60 * 60 * 1000 - 1);
|
||||
expect(failure).toBeUndefined();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await observed;
|
||||
expect(failure).toBeInstanceOf(Error);
|
||||
expect((failure as Error).message).toContain(`native session timed out after ${turnTimeoutMs}ms`);
|
||||
expect(cancel).toHaveBeenCalledOnce();
|
||||
expect(completeRun).not.toHaveBeenCalled();
|
||||
} else {
|
||||
await vi.advanceTimersByTimeAsync(2 * 24 * 60 * 60 * 1000);
|
||||
finish();
|
||||
expect(await observed).toMatchObject({ result });
|
||||
expect(completeRun).toHaveBeenCalledOnce();
|
||||
expect(cancel).not.toHaveBeenCalled();
|
||||
}
|
||||
} finally {
|
||||
finish();
|
||||
vi.useRealTimers();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it.each([false, true])("preserves a durable session failure when its stream closes (throws=%s)", async throws => {
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
const session: NativeSession = {
|
||||
|
|
|
|||
|
|
@ -169,7 +169,10 @@ export interface ExecuteNativeSessionOptions {
|
|||
/** Trusted provider resource identity: independent remote sandboxes must not
|
||||
* inherit each other's process-cleanup gates. Omit for local backends. */
|
||||
remoteCleanupScope?: string;
|
||||
/** Operation bound; explicit values also preserve the legacy turn bound. */
|
||||
timeoutMs?: number;
|
||||
/** Total turn duration. Zero or no configured bound allows long-running work. */
|
||||
turnTimeoutMs?: number;
|
||||
/** Abort admission while waiting for prior cleanup in the same domain. */
|
||||
signal?: AbortSignal;
|
||||
/** Internal test seam; production bounds checkpoint persistence to 30 seconds. */
|
||||
|
|
@ -1158,9 +1161,19 @@ async function consumeTurn(
|
|||
return await Promise.race([
|
||||
consumer,
|
||||
new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(() => {
|
||||
reject(new Error(`native session timed out after ${timeoutMs}ms`));
|
||||
}, timeoutMs);
|
||||
if (timeoutMs <= 0) return;
|
||||
// Node timers overflow above ~24.8 days. Keep explicit long deadlines
|
||||
// in bounded chunks instead of accidentally firing them immediately.
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
const checkDeadline = () => {
|
||||
const remaining = deadline - Date.now();
|
||||
if (remaining <= 0) {
|
||||
reject(new Error(`native session timed out after ${timeoutMs}ms`));
|
||||
} else {
|
||||
timer = setTimeout(checkDeadline, Math.min(remaining, 2_147_483_647));
|
||||
}
|
||||
};
|
||||
checkDeadline();
|
||||
}),
|
||||
handoffFailure,
|
||||
externalAbortFailure,
|
||||
|
|
@ -2233,7 +2246,7 @@ export async function executeNativeSession(
|
|||
session,
|
||||
options.controlPlane,
|
||||
input,
|
||||
options.timeoutMs ?? 900_000,
|
||||
options.turnTimeoutMs ?? options.timeoutMs ?? 0,
|
||||
options.runtimeInputLiveWindowMs ??
|
||||
DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS,
|
||||
options.keepSessionOpen
|
||||
|
|
|
|||
|
|
@ -19,6 +19,8 @@
|
|||
// misdelivering it. `omitHostRouteId: true` sends the notification with no
|
||||
// `hostRouteId` field at all, so a test proves the host warns about a plugin
|
||||
// build old enough to omit the field, instead of silently dropping it.
|
||||
// - `emitOnInput`: when true, wait for the first input before sending scripted
|
||||
// notifications, so legacy routing tests know the worker session is bound.
|
||||
// - `exitCode`: when set, the fixture emits an exit notification after the outputs.
|
||||
// - `omitHostRouteIdOnExit`: when true, the main `exitCode` exit notification
|
||||
// carries no `hostRouteId` field, so a test proves the host still resolves
|
||||
|
|
@ -184,7 +186,13 @@ rl.on("line", (line) => {
|
|||
const mode = directive.mode ?? "normal";
|
||||
const workerSessionId = directive.workerSessionId ?? "ws-1";
|
||||
const closeMode = directive.closeMode ?? "ack";
|
||||
routes.set(params.hostRouteId, { workerSessionId, closeMode });
|
||||
routes.set(params.hostRouteId, {
|
||||
workerSessionId,
|
||||
closeMode,
|
||||
pendingOutput: directive.emitOnInput === true
|
||||
? scriptedOutputLines(directive, params.hostRouteId, workerSessionId)
|
||||
: null,
|
||||
});
|
||||
|
||||
if (mode === "no-open-reply") {
|
||||
// Never reply, so the host open call times out.
|
||||
|
|
@ -230,6 +238,8 @@ rl.on("line", (line) => {
|
|||
return;
|
||||
}
|
||||
|
||||
if (directive.emitOnInput === true) return;
|
||||
|
||||
// Emit the scripted output and the exit after the open reply, so the host
|
||||
// binds the route first.
|
||||
setImmediate(() => {
|
||||
|
|
@ -243,6 +253,11 @@ rl.on("line", (line) => {
|
|||
// test proves the input reaches the worker and the output routes back.
|
||||
for (const [hostRouteId, entry] of routes.entries()) {
|
||||
if (entry.workerSessionId === params.workerSessionId) {
|
||||
if (entry.pendingOutput !== null) {
|
||||
process.stdout.write(entry.pendingOutput);
|
||||
entry.pendingOutput = null;
|
||||
continue;
|
||||
}
|
||||
send({
|
||||
jsonrpc: "2.0",
|
||||
method: "loginPty.output",
|
||||
|
|
|
|||
|
|
@ -1636,10 +1636,13 @@ describe("plugin worker manager login pseudo-terminal missing hostRouteId diagno
|
|||
const route = await handle.openLoginPtySession(
|
||||
ptyOpenInput({
|
||||
workerSessionId: "ws-A",
|
||||
emitOnInput: true,
|
||||
outputs: [{ chunk: "legacy-output", omitHostRouteId: true }],
|
||||
}),
|
||||
);
|
||||
route.onData((chunk) => chunks.push(chunk));
|
||||
// Legacy notifications require the open reply to bind the worker ID.
|
||||
route.write("emit-scripted-output");
|
||||
await vi.waitFor(() => expect(chunks).toContain("legacy-output"));
|
||||
|
||||
const warnCalls = vi.mocked(logger.warn).mock.calls.flat().map((arg) => JSON.stringify(arg));
|
||||
|
|
@ -1661,8 +1664,9 @@ describe("plugin worker manager login pseudo-terminal missing hostRouteId diagno
|
|||
try {
|
||||
await handle.start();
|
||||
const route = await handle.openLoginPtySession(
|
||||
ptyOpenInput({ workerSessionId: "ws-A", exitCode: 0, omitHostRouteIdOnExit: true }),
|
||||
ptyOpenInput({ workerSessionId: "ws-A", exitCode: 0, omitHostRouteIdOnExit: true, emitOnInput: true }),
|
||||
);
|
||||
route.write("emit-scripted-exit");
|
||||
await expect(route.wait()).resolves.toEqual({ exitCode: 0 });
|
||||
} finally {
|
||||
await handle.stop().catch(() => undefined);
|
||||
|
|
|
|||
|
|
@ -4612,7 +4612,9 @@ describe("ensureRuntimeServicesForRun", () => {
|
|||
command: serviceCommand,
|
||||
cwd: ".",
|
||||
port: { type: "auto" as const },
|
||||
readiness: { type: "http" as const, urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 3, intervalMs: 100 },
|
||||
// This checks replacement, not startup latency. Allow the same startup
|
||||
// budget as other real-process fixtures on busy CI hosts.
|
||||
readiness: { type: "http" as const, urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 10, intervalMs: 100 },
|
||||
expose: { type: "url" as const, urlTemplate: "http://127.0.0.1:{{port}}" },
|
||||
lifecycle: "shared" as const,
|
||||
stopPolicy: { type: "manual" as const },
|
||||
|
|
@ -4638,7 +4640,7 @@ describe("ensureRuntimeServicesForRun", () => {
|
|||
});
|
||||
await fs.rm(workspaceRoot, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
}, 30_000);
|
||||
|
||||
it("reuses a shared Paperclip dev runtime after one transient unhealthy response", async () => {
|
||||
const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-transient-health-"));
|
||||
|
|
|
|||
|
|
@ -22916,6 +22916,7 @@ export function heartbeatService(
|
|||
executePaperclipNativeSession({
|
||||
db,
|
||||
execution: nativeExecution,
|
||||
turnTimeoutMs: Math.max(0, asNumber(runtimeConfig.timeoutSec, 0)) * 1_000,
|
||||
runnerInstanceId: nativeRunnerInstanceId,
|
||||
leaseOwner: runOptions.nativeLeaseOwner,
|
||||
restartRecovery: runOptions.nativeRestartRecovery,
|
||||
|
|
|
|||
|
|
@ -5798,26 +5798,31 @@ describe("native warm session supervision", () => {
|
|||
return result;
|
||||
});
|
||||
|
||||
await executePaperclipNativeSession({
|
||||
db: leaseDb(base),
|
||||
execution: base,
|
||||
runnerInstanceId: "runner",
|
||||
});
|
||||
await executePaperclipNativeSession({
|
||||
db: leaseDb(lowered),
|
||||
execution: lowered,
|
||||
runnerInstanceId: "runner",
|
||||
});
|
||||
expect(firstClose).toHaveBeenCalledWith({
|
||||
reason: "warm native session configuration changed",
|
||||
});
|
||||
await vi.waitFor(
|
||||
() =>
|
||||
expect(secondClose).toHaveBeenCalledWith({
|
||||
reason: "warm native session idle timeout",
|
||||
}),
|
||||
{ timeout: 500 },
|
||||
);
|
||||
// Filesystem work between calls can exceed the idle window on a busy host.
|
||||
// Advance that window only after proving the permission change closed it.
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
try {
|
||||
await executePaperclipNativeSession({
|
||||
db: leaseDb(base),
|
||||
execution: base,
|
||||
runnerInstanceId: "runner",
|
||||
});
|
||||
await executePaperclipNativeSession({
|
||||
db: leaseDb(lowered),
|
||||
execution: lowered,
|
||||
runnerInstanceId: "runner",
|
||||
});
|
||||
expect(firstClose).toHaveBeenCalledWith({
|
||||
reason: "warm native session configuration changed",
|
||||
});
|
||||
expect(secondClose).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
expect(secondClose).toHaveBeenCalledWith({
|
||||
reason: "warm native session idle timeout",
|
||||
});
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -6615,6 +6615,8 @@ export async function executePaperclipNativeSession(input: {
|
|||
db: Db;
|
||||
execution: NativeExecutionInput;
|
||||
runnerInstanceId: string;
|
||||
/** Configured total turn bound; zero/unset is unlimited. */
|
||||
turnTimeoutMs?: number;
|
||||
leaseOwner?: string;
|
||||
restartRecovery?: NativeRestartRecoveryClaim;
|
||||
onSpawn?: (meta: {
|
||||
|
|
@ -6635,8 +6637,6 @@ export async function executePaperclipNativeSession(input: {
|
|||
onGoalCheckpoint?: (snapshot: PersistedNativeSession) => Promise<void>;
|
||||
sessionGoalControl?: NativeSessionGoalControl | null;
|
||||
resumeSessionGoalHeartbeat?: boolean;
|
||||
/** Internal test seam; production rolls over five minutes before runnerd's one-hour lease. */
|
||||
goalRolloverAtMs?: number;
|
||||
preparationSpans?: NativeRunHistoricalSpan[];
|
||||
/** Resolved adapter env; the runner transport applies a provider allowlist before spawn. */
|
||||
runnerEnvironment?: NodeJS.ProcessEnv;
|
||||
|
|
@ -7632,6 +7632,7 @@ async function executePaperclipNativeSessionWithinScope(
|
|||
executeNativeSession({
|
||||
input: runnerExecution,
|
||||
remoteCleanupScope: remoteCleanupLease ? remoteLeaseCleanupScope(remoteCleanupLease) : undefined,
|
||||
turnTimeoutMs: input.turnTimeoutMs,
|
||||
backend:
|
||||
input.backend ??
|
||||
runnerdBackend ??
|
||||
|
|
|
|||
Loading…
Reference in New Issue