diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md index 59f7513dfe..e9ef2addb6 100644 --- a/.github/pull_request_template.md +++ b/.github/pull_request_template.md @@ -13,8 +13,8 @@ Replace both values below. Use `feature` or `bugfix` for user-visible behavior changes and add exactly one case under `tests/relayflows/cases//`. Use `non-functional` and `n/a` only when runtime behavior is unchanged. -- Change type: `replace-me` -- RelayFlow case: `replace-me` +- Change type: `feature`, `bugfix`, or `non-functional` +- RelayFlow case: `` or `n/a` for non-functional changes ## Screenshots diff --git a/CHANGELOG.md b/CHANGELOG.md index 948a255b0b..305584a33e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `agent-relay node up` retries the narrowly transient Relaycast `workspace_busy` admission response while keeping unrelated rate limits terminal and preserving bounded startup diagnostics. +- Worker deaths now persist bounded, generation-correlated exit diagnostics immediately and surface them through node and Fleet agent events. Fleet invocation correlation is carried on the exited process generation itself, so a same-name replacement worker can no longer overwrite the exit attribution. Hosted `agent_exited` delivery is now a durable, at-least-once outbox tied to the persisted crash record: a full or closed hosted-event channel no longer silently drops the terminal event, pending deliveries survive a broker restart and are replayed in order (deduped by agent + generation), and pending/backlog/drop counts are queryable via the crash-insights API. The durable record is only marked delivered after Relaycast's HTTP call actually succeeds (with a bounded in-process retry/backoff on timeout or 5xx) — not merely after the event reached the internal publisher queue — so a Relaycast outage no longer permanently loses an exit event; the crash-insights API also reports a `publish_failures_total` count so a stuck outage is observable rather than silently retried forever. A failed write of the crash-insights snapshot itself is now retried automatically on every subsequent maintenance flush (instead of only being logged), with `save_failures_total`/`dirty` and a `retention_pressure_total` counter exposed via the crash-insights API so a transient disk failure or sustained pending-outbox pressure is operator-visible rather than silently lost. The in-memory hosted-delivery queue is now continuously replenished from the durable pending outbox after it overflows its 256-entry bound, within the running broker process — no restart required — without ever re-queuing a delivery that is already in flight. + - `fleet spawn --sandbox` uses the provider-neutral durable profile for explicit Daytona and E2B sandboxes, so their measured resource envelopes are routable while Agent37 retains its heavy profile. - Cloud Daytona Fleet provisioning now requires and returns the exact provider sandbox UUID alongside the stable Cloud sandbox ID, enabling ID-bound inspection and cleanup after interrupted launches. diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 3af6873dc1..dee250286d 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -314,6 +314,10 @@ impl BrokerRuntime { let persist = self.persist; let shutdown = &mut self.shutdown; let crash_insights = &self.crash_insights; + let hosted_agent_exit_backlog_len = self.hosted_agent_exit_backlog.len(); + let hosted_agent_exit_dropped_total = self.hosted_agent_exit_dropped_total; + let hosted_agent_exit_publish_failures_total = + self.hosted_agent_exit_publish_failures_total; match req { ListenApiRequest::Spawn { @@ -674,6 +678,13 @@ impl BrokerRuntime { } if let Some((token, invocation_id, session_ref)) = fleet_registration.take() { + // Carry correlation on this specific generation's + // handle, not just the by-name `fleet_inventory` + // entry: a same-name replacement can overwrite that + // entry before this generation is reaped, which + // would misattribute this invocation id to the + // wrong exit. See maintenance.rs reap. + workers.set_invocation_id(&name, invocation_id.clone()); super::fleet::record_fleet_inventory_agent( fleet_control_tx, fleet_inventory, @@ -2077,7 +2088,35 @@ impl BrokerRuntime { }))); } ListenApiRequest::GetCrashInsights { reply } => { - let _ = reply.send(Ok(crash_insights.to_json())); + // Backward-compatible additions: existing fields from + // `crash_insights.to_json()` are unchanged; `hosted_delivery` + // groups the durable outbox/backlog status a REST client can + // poll instead of (or alongside) directly reading + // `crash-insights.json`'s per-record `hosted_delivery` state. + let mut body = crash_insights.to_json(); + if let Some(object) = body.as_object_mut() { + object.insert( + "hosted_delivery".to_string(), + json!({ + "pending": crash_insights.pending_hosted_deliveries().len(), + "in_memory_backlog_len": hosted_agent_exit_backlog_len, + "in_memory_backlog_cap": super::event_loop::HOSTED_AGENT_EXIT_BACKLOG_CAP, + "dropped_total": hosted_agent_exit_dropped_total, + // Truthful operator status: a record only ever + // counts as `pending` above until Relaycast's + // real HTTP emit is confirmed by + // `run_hosted_agent_event_publisher` — never + // merely because it reached that task's queue. + // `publish_failures_total` counts hosted events + // whose bounded in-process retries were + // exhausted without success; those records + // remain `pending` above and are candidates for + // restart replay, not silently marked delivered. + "publish_failures_total": hosted_agent_exit_publish_failures_total, + }), + ); + } + let _ = reply.send(Ok(body)); } ListenApiRequest::GetDeadLetters { reply } => { let now_ms = unix_timestamp_millis(); diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index 159c9ef9a0..c3f48e0118 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -970,6 +970,17 @@ pub(crate) async fn retry_pending_delivery( { Ok(()) => { if let Some(current) = pending_deliveries.get_mut(delivery_id) { + // `attempts` is a pure observability counter of successful + // handoffs, never the failure-budget gate — only + // `failed_attempts` (reset below) is compared against + // `MAX_DELIVERY_RETRIES` to decide termination. A long-lived + // `Wait`-mode delivery can be handed off successfully far + // more than `MAX_DELIVERY_RETRIES` times while its ack is + // still outstanding (each successful handoff resets the + // failure budget), so capping this counter at the failure + // budget previously misrepresented — and, if ever read as a + // budget check elsewhere, could prematurely treat — a + // perfectly healthy, still-pending delivery as exhausted. current.attempts = current.attempts.saturating_add(1); current.failed_attempts = 0; current.next_retry_at = Instant::now() @@ -985,8 +996,11 @@ pub(crate) async fn retry_pending_delivery( } Err(error) => { let should_fail = if let Some(current) = pending_deliveries.get_mut(delivery_id) { - current.attempts = current.attempts.saturating_add(1); - current.failed_attempts = current.failed_attempts.saturating_add(1); + current.attempts = current.attempts.saturating_add(1).min(MAX_DELIVERY_RETRIES); + current.failed_attempts = current + .failed_attempts + .saturating_add(1) + .min(MAX_DELIVERY_RETRIES); current.next_retry_at = Instant::now() + retry_interval; current.last_error = Some(error.to_string()); current.failed_attempts >= MAX_DELIVERY_RETRIES diff --git a/crates/broker/src/runtime/event_loop.rs b/crates/broker/src/runtime/event_loop.rs index f55943dbad..8869085b02 100644 --- a/crates/broker/src/runtime/event_loop.rs +++ b/crates/broker/src/runtime/event_loop.rs @@ -34,17 +34,433 @@ pub(crate) const RESIZE_OWNER_STALE: Duration = Duration::from_secs(300); /// would instead cost up to N times this much. const SHUTDOWN_RELAYCAST_PHASE_TIMEOUT: Duration = Duration::from_millis(2500); +#[derive(Clone)] pub(crate) struct HostedAgentEvent { pub(super) name: String, pub(super) event_type: String, pub(super) payload: serde_json::Map, pub(super) workspace_id: Option, + /// Idempotency key hosted consumers can use to discard a duplicate + /// delivery. Matches `CrashRecord::dedupe_key` (`agent_name::generation`) + /// for the durable record this event was derived from, so a replay after + /// restart can be recognized as the same logical exit even though it is + /// a distinct in-memory `HostedAgentEvent`. + /// + /// Sent to Relaycast as part of the HTTP payload (see + /// [`run_hosted_agent_event_publisher`]) so a hosted consumer receiving + /// this event can itself discard a duplicate delivery. Relaycast's HTTP + /// API has no server-side idempotency/dedupe key parameter today (it is + /// an external SDK maintained out of this repo), so the *server* cannot + /// yet reject a duplicate `POST` outright — this is documented, + /// deliberately at-least-once behavior, not exactly-once, until a + /// companion change lands server-side. Tracked as + /// AgentWorkforce/relay#1752. + pub(super) dedupe_key: String, +} + +/// Outcome of one publisher attempt (including its bounded in-process +/// retries) to deliver a [`HostedAgentEvent`] to Relaycast, reported back to +/// [`BrokerRuntime`] — the sole owner of [`crate::crash_insights::CrashInsights`] +/// — so *it* decides what the durable record's state should be. +/// +/// This is the real ack boundary the durable outbox is supposed to have: a +/// crash record is only ever marked +/// [`Delivered`](crate::crash_insights::HostedDeliveryState::Delivered) after +/// this outcome reports `success = true`, i.e. after Relaycast actually +/// accepted the HTTP `POST` — never merely because the event was somehow +/// handed to this task's channel. +#[derive(Debug, Clone)] +pub(crate) struct HostedDeliveryOutcome { + pub(super) dedupe_key: String, + pub(super) worker: String, + pub(super) event_type: String, + pub(super) success: bool, +} + +/// Bound on in-process retry attempts for one hosted-agent event before the +/// publisher gives up and reports failure back to [`BrokerRuntime`], leaving +/// the durable crash record `Pending` for a future broker restart to replay. +/// Small and bounded deliberately: this task processes events from a single +/// queue in order, so retrying one event here blocks every later event +/// behind it. A handful of quick attempts absorbs a blip (a single slow or +/// flaky 5xx); anything that outlives them is better handled by the +/// restart-replay fallback than by stalling the whole publisher queue. +const HOSTED_PUBLISH_MAX_ATTEMPTS: u32 = 3; + +/// Base backoff between in-process retry attempts, scaled linearly by +/// attempt number (1x, 2x, ...). Kept short since attempts are already +/// capped by [`HOSTED_PUBLISH_MAX_ATTEMPTS`] and each attempt itself may +/// wait up to the per-attempt HTTP timeout below. +const HOSTED_PUBLISH_RETRY_BASE_DELAY: Duration = Duration::from_millis(200); + +/// Rebuild the hosted `agent_exited` event for a durable crash record. +/// +/// Used both for the live path (immediately after +/// [`crate::crash_insights::CrashInsights::record`]) and for replay on +/// broker restart (see `reload_pending_hosted_agent_exit_backlog`), so the +/// two can never drift: whatever a live exit would have published is +/// bit-for-bit what a restart replays from the durable record. +pub(crate) fn hosted_agent_event_from_crash_record( + record: &crate::crash_insights::CrashRecord, +) -> HostedAgentEvent { + let payload = serde_json::json!({ + "code": record.exit_code, + "signal": record.signal, + "reason": record.exit_reason, + "generation": record.generation, + "workspace_id": record.workspace_id, + "spawn_invocation_id": record.spawn_invocation_id, + "fleet_node_name": record.fleet_node_name, + "became_ready": record.became_ready, + "spawned_at": record.spawned_at, + "ready_at": record.ready_at, + "exited_at": record.exited_at, + }) + .as_object() + .cloned() + .unwrap_or_default(); + HostedAgentEvent { + name: record.agent_name.clone(), + event_type: "agent_exited".to_string(), + payload, + workspace_id: record + .workspace_id + .as_deref() + .map(crate::ids::WorkspaceId::new), + dedupe_key: record.dedupe_key(), + } +} + +/// Rebuild the bounded in-memory retry backlog from durable pending records +/// on broker startup, in original (chronological) order. +/// +/// This is the "replay" half of the durable outbox: [`CrashInsights`] on +/// disk is the source of truth for *what* still needs delivering; this +/// merely reconstructs the in-memory `HostedAgentEvent`s the maintenance +/// tick's normal drain/enqueue path already knows how to retry. Bounded by +/// [`HOSTED_AGENT_EXIT_BACKLOG_CAP`] exactly like the live path, so a broker +/// that comes back after accumulating an enormous pending set still starts +/// with a small, boring backlog rather than an unbounded one — the loudly +/// logged overflow-drop is the same code path live delivery uses. +/// +/// [`CrashInsights`]: crate::crash_insights::CrashInsights +pub(crate) fn reload_pending_hosted_agent_exit_backlog( + crash_insights: &crate::crash_insights::CrashInsights, + dropped_total: &mut u64, +) -> VecDeque { + let mut backlog: VecDeque = crash_insights + .pending_hosted_deliveries() + .into_iter() + .map(hosted_agent_event_from_crash_record) + .collect(); + let replayed = backlog.len(); + enforce_hosted_agent_exit_backlog_cap(&mut backlog, dropped_total); + if replayed > 0 { + tracing::info!( + replayed, + retained = backlog.len(), + "replaying pending hosted agent-exit deliveries from durable crash insights after restart" + ); + } + backlog +} + +/// Trim `backlog` down to [`HOSTED_AGENT_EXIT_BACKLOG_CAP`], dropping the +/// oldest entries first and counting/logging every drop loudly. Shared by +/// the live enqueue path and startup replay so both bound growth identically. +fn enforce_hosted_agent_exit_backlog_cap( + backlog: &mut VecDeque, + dropped_total: &mut u64, +) { + enforce_hosted_agent_exit_backlog_cap_tracked(backlog, dropped_total, None); +} + +/// Same as [`enforce_hosted_agent_exit_backlog_cap`], but also clears the +/// dropped event's dedupe key from `in_flight` (when tracking is in use) so +/// a later [`replenish_hosted_agent_exit_backlog`] pass can pick this +/// still-`Pending`-on-disk record back up instead of it being permanently +/// stuck as "in-flight" for a delivery attempt that was actually discarded. +fn enforce_hosted_agent_exit_backlog_cap_tracked( + backlog: &mut VecDeque, + dropped_total: &mut u64, + mut in_flight: Option<&mut HashSet>, +) { + while backlog.len() > HOSTED_AGENT_EXIT_BACKLOG_CAP { + if let Some(dropped) = backlog.pop_front() { + *dropped_total += 1; + if let Some(in_flight) = in_flight.as_deref_mut() { + in_flight.remove(&dropped.dedupe_key); + } + tracing::error!( + worker = %dropped.name, + event_type = %dropped.event_type, + dropped_total = *dropped_total, + "hosted agent exit backlog overflowed; oldest terminal event was dropped — hosted consumers may miss this exit (durable crash-insights record on disk remains authoritative)" + ); + } + } +} + +/// Upper bound on how many terminal hosted-agent events (e.g. `agent_exited`) +/// are held in [`BrokerRuntime::hosted_agent_exit_backlog`] awaiting a retry +/// once `hosted_agent_event_tx` has capacity again. Bounded so a stuck or +/// permanently closed publisher cannot grow this queue without limit; the +/// oldest entry is dropped (and counted, loudly) once the cap is exceeded. +/// Durable state for these events already lives in crash insights on disk — +/// this backlog exists purely to make *delivery* to hosted consumers durable +/// against transient backpressure, not to be a second source of truth. +pub(crate) const HOSTED_AGENT_EXIT_BACKLOG_CAP: usize = 256; + +/// Rebuild the in-memory backlog from durable pending records after +/// overflow drops, without a broker restart. +/// +/// [`HOSTED_AGENT_EXIT_BACKLOG_CAP`] bounds the in-memory backlog, but the +/// durable pending outbox on disk ([`crate::crash_insights::CrashInsights`]) +/// is not similarly bounded to that cap (see the crate-level storage-tradeoff +/// docs) — a broker that falls far enough behind can accumulate far more +/// pending records than the in-memory backlog can hold at once. Previously +/// the only way to pick those dropped-from-memory (but still `Pending` on +/// disk) records back up was a full broker restart (via +/// [`reload_pending_hosted_agent_exit_backlog`]). This function does the +/// same reconstruction *within* the running process: called periodically +/// (every maintenance tick), it scans current durable pending records and +/// re-enqueues any that are neither already backlogged nor currently +/// in-flight to the publisher, up to the cap. +/// +/// `in_flight` is the guard against duplicate in-flight delivery: every +/// dedupe key handed to the backlog or the publisher channel is tracked +/// there until [`BrokerRuntime::handle_hosted_delivery_outcome`] reports its +/// outcome (or it is dropped from the backlog, which also clears the +/// tracking entry so a still-pending record remains eligible for a later +/// replenishment pass). A record already tracked is skipped here — it is +/// either already queued for retry or already being delivered — so this +/// function can never enqueue a second in-flight copy of the same logical +/// exit. +pub(crate) fn replenish_hosted_agent_exit_backlog( + crash_insights: &crate::crash_insights::CrashInsights, + backlog: &mut VecDeque, + in_flight: &mut HashSet, +) -> usize { + let mut replenished = 0usize; + for record in crash_insights.pending_hosted_deliveries() { + if backlog.len() >= HOSTED_AGENT_EXIT_BACKLOG_CAP { + break; + } + let key = record.dedupe_key(); + if in_flight.contains(&key) { + // Already backlogged or already in-flight to the publisher for + // this exact logical exit — never enqueue a duplicate. + continue; + } + backlog.push_back(hosted_agent_event_from_crash_record(record)); + in_flight.insert(key); + replenished += 1; + } + if replenished > 0 { + tracing::info!( + replenished, + backlog_len = backlog.len(), + "replenished hosted agent-exit backlog from durable pending records \ + after prior overflow drops (no restart required)" + ); + } + replenished +} + +/// Attempt to hand a terminal hosted-agent event (`agent_exited`) to the +/// publisher without blocking maintenance. Unlike a bare `try_send`, a full +/// channel does not silently drop the event: it is held in a small bounded +/// backlog and retried on the next call (typically the next maintenance +/// tick, via [`drain_hosted_agent_exit_backlog_tracked`]) once capacity frees up. +/// +/// A *closed* channel (publisher task gone) can never succeed, so the event +/// is not backlogged in that case — only counted and logged at error level so +/// the miss is observable — and the caller's crash-insights persistence +/// (already durable on disk) remains the authoritative record either way. +/// Mark `event` delivered in the durable crash-insights record it was +/// derived from and persist that transition immediately. +/// +/// This is the real "ack boundary" for hosted delivery: it must only be +/// called once [`run_hosted_agent_event_publisher`] reports a +/// [`HostedDeliveryOutcome`] with `success = true`, i.e. after Relaycast's +/// HTTP endpoint actually accepted the event — never merely because the +/// event was handed to the publisher's channel (a successful `try_send` +/// only proves the publisher task received it, not that Relaycast did; an +/// HTTP timeout or 5xx after that handoff must leave the record `Pending`). +/// Persisting right here, synchronously, on the confirmed-success path means +/// a crash immediately after this call still leaves the on-disk record +/// correctly marked `Delivered` — and a crash at any point before it +/// (including while Relaycast's own HTTP call is still in flight, retrying, +/// or has permanently failed) leaves the record `Pending`, so a restart +/// replays it again. That replay is safe (not silently duplicate-lossy) +/// only because consumers dedupe on `dedupe_key`; this is the accepted +/// at-least-once tradeoff. +pub(crate) fn mark_hosted_delivered_and_persist( + crash_insights: &mut crate::crash_insights::CrashInsights, + crash_insights_path: &std::path::Path, + persist: bool, + dedupe_key: &str, +) { + if crash_insights.mark_hosted_delivered(dedupe_key) + && persist + && !crash_insights.persist(crash_insights_path) + { + // `persist` re-armed the dirty flag and bumped the durability + // failure counter; `flush_persisted_stores` retries on the next + // tick, so this is not a permanent loss. Until it succeeds, the + // on-disk record still says `Pending`, so a restart in the + // meantime would harmlessly replay this already-delivered exit + // (safe: hosted consumers dedupe on `dedupe_key`). + tracing::warn!( + path = %crash_insights_path.display(), + dedupe_key = %dedupe_key, + "failed to persist hosted-delivery acknowledgement; will retry on next maintenance flush (a restart in the meantime may harmlessly replay this already-delivered exit)" + ); + } +} + +/// Attempt to hand `event` to the hosted publisher's channel. This is only a +/// *handoff*, not delivery: the durable crash record stays `Pending` even +/// after a successful `try_send` here, because Relaycast's real HTTP call +/// happens later, in [`run_hosted_agent_event_publisher`]. That task reports +/// the true outcome back over its own result channel, and only that outcome +/// (see [`mark_hosted_delivered_and_persist`]) may mark the record +/// `Delivered`. This function's job is purely to make sure a momentarily +/// full or permanently closed channel doesn't cause the event to be +/// silently dropped short of ever reaching the publisher at all. +/// Maintains the +/// `in_flight` dedupe-key set that [`replenish_hosted_agent_exit_backlog`] +/// relies on to avoid ever queuing a second in-memory copy of the same +/// logical exit while one is already backlogged or handed to the publisher. +pub(crate) fn enqueue_hosted_agent_exit_event_tracked( + tx: &mpsc::Sender, + backlog: &mut VecDeque, + dropped_total: &mut u64, + mut in_flight: Option<&mut HashSet>, + event: HostedAgentEvent, +) { + drain_hosted_agent_exit_backlog_tracked(tx, backlog, dropped_total, in_flight.as_deref_mut()); + if let Some(in_flight) = in_flight.as_deref_mut() { + in_flight.insert(event.dedupe_key.clone()); + } + match tx.try_send(event) { + Ok(()) => { + // Handed to the publisher task only — the durable record + // remains `Pending` until that task's real HTTP call to + // Relaycast succeeds and reports back over the delivery-result + // channel. See `BrokerRuntime::handle_hosted_delivery_outcome`. + } + Err(mpsc::error::TrySendError::Full(event)) => { + tracing::warn!( + worker = %event.name, + event_type = %event.event_type, + backlog_len = backlog.len(), + "hosted agent event queue full; holding terminal event in bounded backlog for retry" + ); + backlog.push_back(event); + enforce_hosted_agent_exit_backlog_cap_tracked(backlog, dropped_total, in_flight); + } + Err(mpsc::error::TrySendError::Closed(event)) => { + *dropped_total += 1; + if let Some(in_flight) = in_flight { + in_flight.remove(&event.dedupe_key); + } + tracing::error!( + worker = %event.name, + event_type = %event.event_type, + dropped_total = *dropped_total, + "hosted agent event publisher channel is closed; terminal event could not be delivered to hosted consumers for the remaining process lifetime (durable crash-insights record on disk remains authoritative and pending — it will be replayed on the next broker restart)" + ); + } + } } +/// Retry every backlogged terminal hosted-agent event against the publisher +/// channel, in original order, stopping at the first one that still does not +/// fit so relative ordering is preserved and this call cannot itself stall +/// maintenance on a persistently full channel (bounded work per tick, never +/// blocking maintenance indefinitely on a stuck or closed channel). +/// Clears dropped +/// events' dedupe keys from `in_flight` — see +/// [`enqueue_hosted_agent_exit_event_tracked`] and +/// [`replenish_hosted_agent_exit_backlog`]. +pub(crate) fn drain_hosted_agent_exit_backlog_tracked( + tx: &mpsc::Sender, + backlog: &mut VecDeque, + dropped_total: &mut u64, + mut in_flight: Option<&mut HashSet>, +) { + while let Some(event) = backlog.pop_front() { + match tx.try_send(event) { + Ok(()) => { + // Handoff only — see `enqueue_hosted_agent_exit_event_tracked`. The + // durable record stays `Pending` until the publisher's real + // HTTP call confirms success. Already tracked in `in_flight` + // since the moment it was backlogged/enqueued. + } + Err(mpsc::error::TrySendError::Full(event)) => { + backlog.push_front(event); + break; + } + Err(mpsc::error::TrySendError::Closed(event)) => { + *dropped_total += 1; + if let Some(in_flight) = in_flight.as_deref_mut() { + in_flight.remove(&event.dedupe_key); + } + tracing::error!( + worker = %event.name, + event_type = %event.event_type, + dropped_total = *dropped_total, + "hosted agent event publisher channel closed while draining backlog; terminal event could not be delivered for the remaining process lifetime (durable crash-insights record on disk remains authoritative and pending — it will be replayed on the next broker restart)" + ); + // The channel cannot recover for the rest of this process's + // lifetime; draining further entries would only repeat the + // same closed-channel error for each of them. Report them + // all now rather than one per future tick. Every one of + // these events' durable crash records is still `Pending` on + // disk (never marked delivered) so a broker restart replays + // them — this in-memory backlog is only ever a delivery + // *retry* aid, not the durability boundary itself. + for remaining in backlog.drain(..) { + *dropped_total += 1; + if let Some(in_flight) = in_flight.as_deref_mut() { + in_flight.remove(&remaining.dedupe_key); + } + tracing::error!( + worker = %remaining.name, + event_type = %remaining.event_type, + dropped_total = *dropped_total, + "hosted agent event publisher channel closed; discarding backlogged terminal event from the in-memory retry queue (durable crash-insights record remains pending for replay on restart)" + ); + } + break; + } + } + } +} + +/// Drive the real Relaycast HTTP emit for every hosted-agent event, with a +/// small bounded number of same-process retries on failure (timeout or +/// non-2xx), and report the true outcome back over `result_tx`. +/// +/// This is the actual delivery boundary: [`BrokerRuntime`] must not treat an +/// event as delivered until this task reports `success = true` here — a +/// successful handoff into this task's `rx` channel only proves the event +/// reached this task, not that Relaycast accepted it. See +/// [`HostedDeliveryOutcome`] and [`mark_hosted_delivered_and_persist`]. +/// +/// Retries are deliberately bounded and processed in-line (this task +/// consumes `rx` in order, so a stuck event's retries block later events +/// behind it in the queue). Anything that exhausts +/// [`HOSTED_PUBLISH_MAX_ATTEMPTS`] is reported as a failure so the durable +/// crash record is left `Pending` for the broker-restart replay fallback +/// (see `reload_pending_hosted_agent_exit_backlog`) rather than stalling +/// this queue indefinitely. pub(crate) async fn run_hosted_agent_event_publisher( default_client: RelaycastHttpClient, clients: HashMap, mut rx: mpsc::Receiver, + result_tx: mpsc::Sender, ) { while let Some(event) = rx.recv().await { let client = event @@ -52,15 +468,75 @@ pub(crate) async fn run_hosted_agent_event_publisher( .as_ref() .and_then(|workspace_id| clients.get(workspace_id)) .unwrap_or(&default_client); - if let Err(error) = tokio::time::timeout( - Duration::from_secs(5), - client.emit_agent_event(&event.name, event.event_type, event.payload), - ) - .await - .map_err(|_| anyhow::anyhow!("Relaycast agent event publish timed out")) - .and_then(|result| result) + + // Carry the dedupe identity into the actual HTTP payload sent to + // Relaycast so a hosted consumer receiving this event — including a + // replayed one after a broker restart — can discard a duplicate + // itself. Relaycast's HTTP API has no server-side idempotency-key + // parameter today (it's an external SDK maintained out of this + // repo), so this is client-observable dedupe support only; delivery + // remains documented at-least-once, not exactly-once, until a + // companion server-side change lands (AgentWorkforce/relay#1752). + let mut payload = event.payload.clone(); + payload + .entry("dedupe_key".to_string()) + .or_insert_with(|| Value::String(event.dedupe_key.clone())); + + let mut attempt = 0u32; + let success = loop { + attempt += 1; + let outcome = tokio::time::timeout( + Duration::from_secs(5), + client.emit_agent_event(&event.name, event.event_type.clone(), payload.clone()), + ) + .await + .map_err(|_| anyhow::anyhow!("Relaycast agent event publish timed out")) + .and_then(|result| result); + + match outcome { + Ok(()) => break true, + Err(error) => { + tracing::warn!( + worker = %event.name, + error = %error, + attempt, + max_attempts = HOSTED_PUBLISH_MAX_ATTEMPTS, + dedupe_key = %event.dedupe_key, + "failed to publish agent event to Relaycast" + ); + if attempt >= HOSTED_PUBLISH_MAX_ATTEMPTS { + break false; + } + tokio::time::sleep(HOSTED_PUBLISH_RETRY_BASE_DELAY * attempt).await; + } + } + }; + + if !success { + tracing::error!( + worker = %event.name, + event_type = %event.event_type, + dedupe_key = %event.dedupe_key, + attempts = attempt, + "exhausted in-process retries publishing hosted agent event to Relaycast; durable crash record remains pending for replay on the next broker restart" + ); + } + + if result_tx + .send(HostedDeliveryOutcome { + dedupe_key: event.dedupe_key.clone(), + worker: event.name.clone(), + event_type: event.event_type.clone(), + success, + }) + .await + .is_err() { - tracing::warn!(worker = %event.name, error = %error, "failed to publish agent event to Relaycast"); + tracing::warn!( + worker = %event.name, + dedupe_key = %event.dedupe_key, + "hosted delivery outcome channel closed; durable crash-insights record's fate now relies solely on restart replay" + ); } } } @@ -203,6 +679,39 @@ pub(crate) struct BrokerRuntime { pub(super) ws_control_tx: mpsc::Sender, pub(super) relaycast_http: RelaycastHttpClient, pub(super) hosted_agent_event_tx: mpsc::Sender, + /// Bounded retry buffer for terminal hosted-agent events (currently + /// `agent_exited`) that could not be handed to the publisher because its + /// channel was momentarily full. See [`enqueue_hosted_agent_exit_event_tracked`]. + pub(super) hosted_agent_exit_backlog: VecDeque, + /// Dedupe keys of hosted-agent events currently backlogged or handed to + /// the publisher's channel awaiting a [`HostedDeliveryOutcome`]. Guards + /// [`replenish_hosted_agent_exit_backlog`] against ever queuing a + /// second in-memory copy of the same logical exit while one delivery + /// attempt is already outstanding. Cleared for a key once its outcome + /// is reported (success or failure — see + /// `BrokerRuntime::handle_hosted_delivery_outcome`) or once the event is + /// dropped from the backlog (overflow or closed channel), at which + /// point it becomes eligible for replenishment again if — and only if — + /// its durable record is still `Pending`. + pub(super) hosted_agent_exit_in_flight: HashSet, + /// Count of terminal hosted-agent events that were ultimately never + /// delivered to hosted consumers (backlog overflow or a closed + /// publisher channel). Observable via logs at error level; kept here so + /// tests can assert on it directly. + pub(super) hosted_agent_exit_dropped_total: u64, + /// Receives the true delivery outcome of every hosted-agent event from + /// [`run_hosted_agent_event_publisher`] — the only place a durable crash + /// record may be marked `Delivered` (see + /// [`BrokerRuntime::handle_hosted_delivery_outcome`]). + pub(super) hosted_delivery_result_rx: mpsc::Receiver, + pub(super) hosted_delivery_result_open: bool, + /// Count of hosted-agent events whose publisher exhausted in-process + /// retries (see [`HOSTED_PUBLISH_MAX_ATTEMPTS`]) without success. These + /// records remain `Pending` on disk and are candidates for restart + /// replay; this counter is purely observability (truthful operator + /// status), exposed alongside `hosted_delivery_pending` in + /// `GetCrashInsights`. + pub(super) hosted_agent_exit_publish_failures_total: u64, pub(super) pty_observability: HashMap, pub(super) api_rx: mpsc::Receiver, pub(super) api_open: bool, @@ -290,6 +799,7 @@ enum RuntimeEvent { Fleet(Option), Terminal(Option), Worker(Option), + HostedDeliveryResult(Option), MaintenanceTick, } @@ -335,6 +845,7 @@ impl BrokerRuntime { event = self.fleet_event_rx.recv(), if self.fleet_control_open => RuntimeEvent::Fleet(event), event = self.terminal_event_rx.recv(), if self.terminal_control_open => RuntimeEvent::Terminal(event), event = self.worker_event_rx.recv(), if self.worker_events_open => RuntimeEvent::Worker(event), + outcome = self.hosted_delivery_result_rx.recv(), if self.hosted_delivery_result_open => RuntimeEvent::HostedDeliveryResult(outcome), _ = self.reap_tick.tick() => RuntimeEvent::MaintenanceTick, }; @@ -384,6 +895,12 @@ impl BrokerRuntime { RuntimeEvent::Worker(None) => { self.worker_events_open = false; } + RuntimeEvent::HostedDeliveryResult(Some(outcome)) => { + self.handle_hosted_delivery_outcome(outcome); + } + RuntimeEvent::HostedDeliveryResult(None) => { + self.hosted_delivery_result_open = false; + } RuntimeEvent::MaintenanceTick => { self.handle_maintenance_tick().await; } @@ -446,6 +963,57 @@ impl BrokerRuntime { self.dedup.mark_dirty(); } } + // Guaranteed-retry flush for crash insights (the durable + // hosted-delivery outbox), mirroring the dirty-flag pattern above. + // Every mutation (`record`, `mark_hosted_delivered`) — live or via a + // prior failed `persist` call re-arming the flag — sets `dirty`; + // this runs every event-loop iteration (not just maintenance ticks) + // so a transient write failure gets retried promptly rather than + // waiting for the next reap interval. + if self.crash_insights.take_dirty() + && !self.crash_insights.persist(&self.crash_insights_path) + { + tracing::warn!( + path = %self.crash_insights_path.display(), + save_failures_total = self.crash_insights.save_failures_total(), + "failed to persist crash insights — will retry on next flush" + ); + } + } + + /// Apply the real outcome of a hosted-agent event's Relaycast HTTP + /// publish, reported by [`run_hosted_agent_event_publisher`]. This is + /// the only place a durable crash record's `hosted_delivery` field may + /// transition to `Delivered` — success here means Relaycast actually + /// accepted the event, not merely that it reached the publisher's + /// channel. On failure the record is deliberately left `Pending` + /// (already its state — this function is a no-op on the durable record + /// in that case) so a future broker restart replays it; the only action + /// taken is bumping the observability counter so operator status + /// reflects the real failure rather than a false `Delivered`. + fn handle_hosted_delivery_outcome(&mut self, outcome: HostedDeliveryOutcome) { + // This delivery attempt is resolved either way (delivered, or + // failed and left `Pending` for replay) — it is no longer + // in-flight, so a still-pending record becomes eligible for + // `replenish_hosted_agent_exit_backlog` again on the next tick. + self.hosted_agent_exit_in_flight.remove(&outcome.dedupe_key); + if outcome.success { + mark_hosted_delivered_and_persist( + &mut self.crash_insights, + &self.crash_insights_path, + self.paths.persist, + &outcome.dedupe_key, + ); + } else { + self.hosted_agent_exit_publish_failures_total += 1; + tracing::error!( + worker = %outcome.worker, + event_type = %outcome.event_type, + dedupe_key = %outcome.dedupe_key, + publish_failures_total = self.hosted_agent_exit_publish_failures_total, + "hosted agent event publish to Relaycast failed after in-process retries; durable crash record remains pending for restart replay" + ); + } } fn handle_lease_tick(&mut self) { @@ -464,10 +1032,11 @@ impl BrokerRuntime { async fn shutdown_runtime(mut self) -> Result<()> { self.drain_identity_cleanups_on_shutdown().await; // Save crash insights before shutdown (only in persist mode) - if self.paths.persist { - if let Err(error) = self.crash_insights.save(&self.crash_insights_path) { - tracing::warn!(error = %error, "failed to save crash insights"); - } + if self.paths.persist && !self.crash_insights.persist(&self.crash_insights_path) { + tracing::warn!( + save_failures_total = self.crash_insights.save_failures_total(), + "failed to save crash insights on shutdown" + ); } self.telemetry.track(TelemetryEvent::BrokerStop { @@ -602,25 +1171,157 @@ mod resize_owner_tests { ); let workspace_id = WorkspaceId::new("ws_secondary"); let (tx, rx) = mpsc::channel(4); + let (result_tx, mut result_rx) = mpsc::channel(4); let task = tokio::spawn(run_hosted_agent_event_publisher( default_client, HashMap::from([(workspace_id.clone(), secondary_client)]), rx, + result_tx, )); tx.send(HostedAgentEvent { name: "Worker".to_string(), event_type: "activity.changed".to_string(), payload: serde_json::Map::new(), workspace_id: Some(workspace_id), + dedupe_key: "Worker::activity.changed".to_string(), }) .await .expect("publisher queue open"); + let outcome = result_rx + .recv() + .await + .expect("publisher must report a delivery outcome"); + assert!(outcome.success, "successful HTTP emit must report success"); + assert_eq!(outcome.dedupe_key, "Worker::activity.changed"); drop(tx); task.await.expect("publisher task"); secondary_post.assert_hits(1); default_post.assert_hits(0); } + /// Critical fix for #1603/#1750 review: an HTTP timeout or 5xx from + /// Relaycast must not leave the durable crash record silently marked + /// `Delivered`. This drives the publisher directly against a mock + /// server that always fails, asserting the reported outcome is + /// `success = false` and that the bounded in-process retry actually + /// attempted more than once before giving up. + #[tokio::test] + async fn publisher_reports_failure_after_exhausting_retries_on_persistent_5xx() { + let server = MockServer::start(); + let failing = server.mock(|when, then| { + when.method(POST).path("/v1/agents/Worker/events"); + then.status(500) + .json_body(json!({"ok": false, "error": "boom"})); + }); + let client = + RelaycastHttpClient::new(Some(server.base_url()), "rk_test", "broker", "codex"); + let (tx, rx) = mpsc::channel(4); + let (result_tx, mut result_rx) = mpsc::channel(4); + let task = tokio::spawn(run_hosted_agent_event_publisher( + client, + HashMap::new(), + rx, + result_tx, + )); + tx.send(HostedAgentEvent { + name: "Worker".to_string(), + event_type: "agent_exited".to_string(), + payload: serde_json::Map::new(), + workspace_id: None, + dedupe_key: "Worker::gen-1".to_string(), + }) + .await + .expect("publisher queue open"); + + let outcome = result_rx + .recv() + .await + .expect("publisher must report an outcome even on total failure"); + assert!( + !outcome.success, + "persistent 5xx must be reported as failure, never as a false success" + ); + assert_eq!(outcome.dedupe_key, "Worker::gen-1"); + drop(tx); + task.await.expect("publisher task"); + failing.assert_hits(HOSTED_PUBLISH_MAX_ATTEMPTS as usize); + } + + /// The success half of the same scenario: Relaycast 5xxs on the first + /// attempt but recovers within the bounded retry budget — the publisher + /// must report `success = true` only once the real HTTP call actually + /// went through, and the failing mock must have been hit at least once + /// (proving an in-process retry actually happened, not a first-try + /// fluke). + #[tokio::test] + async fn publisher_recovers_and_reports_success_after_transient_5xx() { + let server = MockServer::start(); + let mut failing = server.mock(|when, then| { + when.method(POST).path("/v1/agents/Worker/events"); + then.status(503) + .json_body(json!({"ok": false, "error": "unavailable"})); + }); + let client = + RelaycastHttpClient::new(Some(server.base_url()), "rk_test", "broker", "codex"); + let (tx, rx) = mpsc::channel(4); + let (result_tx, mut result_rx) = mpsc::channel(4); + let task = tokio::spawn(run_hosted_agent_event_publisher( + client, + HashMap::new(), + rx, + result_tx, + )); + + // Once the first (failing) attempt lands, swap the mock for a + // success response so the bounded retry's next attempt recovers — + // driven concurrently with the publisher awaiting its retry backoff. + let recover = async { + while failing.hits() == 0 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + failing.delete(); + server.mock(|when, then| { + when.method(POST).path("/v1/agents/Worker/events"); + then.status(200).json_body(json!({"ok":true,"data":{"id":"evt_recovered","agent_id":"a","type":"agent_exited","payload":{},"created_at":"2026-07-16T00:00:00Z"}})); + }); + }; + + let send = async { + tx.send(HostedAgentEvent { + name: "Worker".to_string(), + event_type: "agent_exited".to_string(), + payload: serde_json::Map::new(), + workspace_id: None, + dedupe_key: "Worker::gen-2".to_string(), + }) + .await + .expect("publisher queue open"); + }; + + let (outcome, _, ()) = tokio::time::timeout(Duration::from_secs(10), async { + tokio::join!( + async { + result_rx + .recv() + .await + .expect("publisher must report an outcome") + }, + recover, + send, + ) + }) + .await + .expect("publisher recovery must complete within the retry budget"); + + assert!( + outcome.success, + "recovery within the bounded retry budget must be reported as success" + ); + assert_eq!(outcome.dedupe_key, "Worker::gen-2"); + drop(tx); + task.await.expect("publisher task"); + } + #[test] fn legacy_request_without_session_always_applies() { // Someone else owns it, but a request with no session id still applies. diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index a8b3323fd0..0329eab155 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2763,6 +2763,7 @@ mod tests { context_budget_pct: None, state: crate::worker::AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); diff --git a/crates/broker/src/runtime/init.rs b/crates/broker/src/runtime/init.rs index 12058b3864..ac3897e714 100644 --- a/crates/broker/src/runtime/init.rs +++ b/crates/broker/src/runtime/init.rs @@ -219,6 +219,16 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re let ws_control_tx = default_workspace.ws_control_tx.clone(); let relaycast_http = default_workspace.http_client.clone(); let (hosted_agent_event_tx, hosted_agent_event_rx) = mpsc::channel::(10_000); + // Delivery-result channel: the publisher task reports the real outcome + // of every Relaycast HTTP emit back here so `BrokerRuntime` — the sole + // owner of `CrashInsights` — is the only thing that ever marks a durable + // crash record `Delivered`. Sized generously relative to the event + // channel above; a full result channel would only cause a warning log in + // the publisher (see `run_hosted_agent_event_publisher`), never data loss + // for the durable record itself, since an unconfirmed record simply + // stays `Pending` and is replayed on restart. + let (hosted_delivery_result_tx, hosted_delivery_result_rx) = + mpsc::channel::(10_000); let hosted_event_client = relaycast_http.clone(); let hosted_event_clients = workspaces .iter() @@ -233,6 +243,7 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re hosted_event_client, hosted_event_clients, hosted_agent_event_rx, + hosted_delivery_result_tx, )); let node_workspace_id = default_workspace.workspace_id.as_str().to_string(); let node_id = resolve_broker_node_id(&node_workspace_id); @@ -556,6 +567,27 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re // Load crash insights from previous session let crash_insights_path = paths.state.parent().unwrap().join("crash-insights.json"); let crash_insights = crate::crash_insights::CrashInsights::load(&crash_insights_path); + // Replay the durable hosted-delivery outbox: any exit whose + // `agent_exited` hand-off to the hosted publisher channel never + // succeeded (crash before or during that handoff, or a channel that was + // closed for the remainder of the previous process's life) is still + // `Pending` on disk and gets a fresh in-memory retry entry here, in the + // same order it was recorded, so this restart's normal maintenance-tick + // drain path retries it exactly as it would a same-session backlog + // entry. See `crate::crash_insights::CrashRecord::hosted_delivery`. + let mut hosted_agent_exit_dropped_total = 0u64; + let hosted_agent_exit_backlog = super::event_loop::reload_pending_hosted_agent_exit_backlog( + &crash_insights, + &mut hosted_agent_exit_dropped_total, + ); + // Seed the in-flight dedupe-key guard with every record just replayed + // into the backlog above, so this restart's first + // `replenish_hosted_agent_exit_backlog` pass (see the maintenance tick) + // does not immediately re-enqueue a second copy of any of them. + let hosted_agent_exit_in_flight: std::collections::HashSet = hosted_agent_exit_backlog + .iter() + .map(|event| event.dedupe_key.clone()) + .collect(); let sdk_lines = BufReader::new(tokio::io::stdin()).lines(); let stdin_open = true; @@ -674,6 +706,12 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re ws_control_tx, relaycast_http, hosted_agent_event_tx, + hosted_agent_exit_backlog, + hosted_agent_exit_in_flight, + hosted_agent_exit_dropped_total, + hosted_delivery_result_rx, + hosted_delivery_result_open: true, + hosted_agent_exit_publish_failures_total: 0, pty_observability: HashMap::new(), api_rx, api_open: true, diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 2aa92224d6..714ab58dce 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -2,6 +2,49 @@ use super::fleet::{release_terminal_resize_ownership, try_send_terminal}; use super::*; use crate::terminal_control::TerminalToCloud; +fn instant_to_unix_secs(at: Instant, now: Instant, now_unix: u64) -> u64 { + now_unix.saturating_sub(now.saturating_duration_since(at).as_secs()) +} + +fn bounded_exit_reason(reason: Option<&str>) -> Option { + reason.map(|reason| { + const MAX_BYTES: usize = 256; + if reason.len() <= MAX_BYTES { + reason.to_string() + } else { + let end = reason.floor_char_boundary(MAX_BYTES - '…'.len_utf8()); + format!("{}…", &reason[..end]) + } + }) +} + +#[cfg(test)] +mod exit_diagnostic_tests { + use super::*; + + #[test] + fn exit_reason_is_bounded_without_splitting_utf8() { + let reason = "é".repeat(300); + let bounded = bounded_exit_reason(Some(&reason)).expect("reason is present"); + assert!(bounded.len() <= 256); + assert!(bounded.ends_with('…')); + assert!(std::str::from_utf8(bounded.as_bytes()).is_ok()); + } + + #[test] + fn instant_conversion_is_monotonic_and_bounded() { + let now = Instant::now(); + let spawned = now - Duration::from_secs(12); + let ready = now - Duration::from_secs(5); + assert_eq!(instant_to_unix_secs(spawned, now, 1_000), 988); + assert_eq!(instant_to_unix_secs(ready, now, 1_000), 995); + assert_eq!( + instant_to_unix_secs(now + Duration::from_secs(1), now, 1_000), + 1_000 + ); + } +} + impl BrokerRuntime { pub(super) async fn handle_maintenance_tick(&mut self) { self.reconcile_identity_cleanups().await; @@ -11,6 +54,9 @@ impl BrokerRuntime { let ws_control_tx = &self.ws_control_tx; let relaycast_http = &self.relaycast_http; let hosted_agent_event_tx = &self.hosted_agent_event_tx; + let hosted_agent_exit_backlog = &mut self.hosted_agent_exit_backlog; + let hosted_agent_exit_dropped_total = &mut self.hosted_agent_exit_dropped_total; + let hosted_agent_exit_in_flight = &mut self.hosted_agent_exit_in_flight; let pty_observability = &mut self.pty_observability; let workers = &mut self.workers; let fleet_control_tx = &self.fleet_control_tx; @@ -37,9 +83,35 @@ impl BrokerRuntime { let delivery_retry_interval = self.delivery_retry_interval; let shutdown = &self.shutdown; let default_workspace = &self.default_workspace; + let fleet_node_name = self.fleet_node_name.clone(); + let crash_insights_path = &self.crash_insights_path; let now = Instant::now(); + // Retry any terminal hosted-agent events (e.g. `agent_exited`) that a + // previous tick could not hand to the publisher because its channel + // was momentarily full. Doing this before generating any new events + // this tick preserves delivery order. + super::event_loop::drain_hosted_agent_exit_backlog_tracked( + hosted_agent_event_tx, + hosted_agent_exit_backlog, + hosted_agent_exit_dropped_total, + Some(hosted_agent_exit_in_flight), + ); + + // Continuously replenish the in-memory backlog from durable pending + // records so overflow drops above `HOSTED_AGENT_EXIT_BACKLOG_CAP` + // are not permanent for the life of the process — no broker restart + // required. `hosted_agent_exit_in_flight` guards against ever + // re-queuing a record that is already backlogged or already + // in-flight to the publisher, so this can never produce a duplicate + // in-flight delivery. + super::event_loop::replenish_hosted_agent_exit_backlog( + crash_insights, + hosted_agent_exit_backlog, + hosted_agent_exit_in_flight, + ); + // A worker can disappear before answering `snapshot_pty`. Bound these // terminal-only RPCs so their sessions cannot remain live forever. let expired_terminal_snapshots: Vec<(String, String, Option)> = @@ -276,8 +348,42 @@ impl BrokerRuntime { vec![] } }; + // Use the instant immediately surrounding reaping for wall-clock + // conversion; delivery/reconciliation awaits earlier in this tick + // must not make the terminal timestamp stale. + let reaped_at = Instant::now(); + let reaped_at_unix = unix_timestamp_secs(); let mut fleet_load_changed = !expired_verified_spawns.is_empty() || !exited.is_empty(); - for (name, generation, code, signal, exit_reason) in &exited { + for ( + name, + generation, + code, + signal, + exit_reason, + workspace_id, + spawned_at, + ready_at, + generation_invocation_id, + ) in &exited + { + // Correlation is carried directly on the exited generation's + // handle (captured by `reap_exited` before the handle was + // removed), not re-derived from the by-name `fleet_inventory` + // map here. A same-name replacement worker can register and + // overwrite that by-name entry before this older generation is + // reaped, which would otherwise misattribute the *new* + // generation's invocation id to *this* (old) generation's exit. + // Fall back to the by-name lookup only for legacy handles that + // never carried an `invocation_id` (e.g. pre-upgrade in-flight + // workers), preserving prior behavior for that narrow case. + let spawn_invocation_id = generation_invocation_id.clone().or_else(|| { + fleet_inventory + .get(name) + .and_then(|agent| agent.invocation_id.clone()) + }); + let was_pending_verified_spawn = pending_verified_spawns + .get(name) + .is_some_and(|pending| pending.generation == *generation); let mut retain_fleet_identity = workers .owned_spawn_generations .get(name) @@ -334,7 +440,27 @@ impl BrokerRuntime { .await; } } - let lifecycle_reason = exit_reason.as_deref().unwrap_or("worker_exited"); + let exited_at = reaped_at_unix; + let spawned_at_unix = instant_to_unix_secs(*spawned_at, reaped_at, reaped_at_unix); + let ready_at_unix = + ready_at.map(|at| instant_to_unix_secs(at, reaped_at, reaped_at_unix)); + let (category, description) = + crate::crash_insights::CrashInsights::analyze(*code, signal.as_deref()); + let durable_reason = bounded_exit_reason(exit_reason.as_deref()) + .or_else(|| { + was_pending_verified_spawn.then(|| "spawn_harness_not_ready".to_string()) + }) + .or_else(|| { + if *code == Some(0) && signal.is_none() { + Some("clean_exit".to_string()) + } else { + Some(description.clone()) + } + }); + let generation_id = generation.to_string(); + let workspace_id = workspace_id.as_ref().map(ToString::to_string); + let fleet_node_name = (!fleet_node_name.is_empty()).then(|| fleet_node_name.clone()); + let lifecycle_reason = durable_reason.as_deref().unwrap_or("worker_exited"); if (code.is_some_and(|code| code != 0) || signal.is_some()) && state .agents @@ -370,21 +496,63 @@ impl BrokerRuntime { tracing::warn!(target = "relay_broker::terminal", session_id = %session_id, "terminal queue full or closed while closing exited worker session"); } } - // Record crash in insights - let (category, description) = - crate::crash_insights::CrashInsights::analyze(*code, signal.as_deref()); - crash_insights.record(crate::crash_insights::CrashRecord { + // Record and persist the terminal crash record — with hosted + // delivery defaulted to `Pending` — *before* any attempt to hand + // it to the hosted publisher channel below. This ordering is the + // core of the durable outbox: a broker crash between this save + // and the enqueue attempt still leaves a `Pending` record on + // disk, so a restart replays it; there is no window in which the + // event is neither durably queued nor delivered. + let crash_record = crate::crash_insights::CrashRecord { agent_name: name.as_str().to_string(), exit_code: *code, signal: signal.clone(), - timestamp: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - uptime_secs: 0, + timestamp: exited_at, + uptime_secs: exited_at.saturating_sub(spawned_at_unix), category, - description, - }); + description: description.clone(), + workspace_id: workspace_id.clone(), + spawn_invocation_id: spawn_invocation_id.clone(), + generation: generation_id.clone(), + became_ready: ready_at.is_some(), + spawned_at: spawned_at_unix, + ready_at: ready_at_unix, + exited_at, + exit_reason: durable_reason.clone(), + fleet_node_name: fleet_node_name.clone(), + hosted_delivery: crate::crash_insights::HostedDeliveryState::Pending, + }; + crash_insights.record(crash_record.clone()); + if paths.persist && !crash_insights.persist(crash_insights_path) { + // `persist` already re-armed the dirty flag and bumped + // `save_failures_total` (surfaced via `GetCrashInsights`); + // the next maintenance tick's flush + // (`BrokerRuntime::flush_persisted_stores`) retries this + // write automatically, so a transient failure here cannot + // silently and permanently lose this exit's durability — + // only the logged warning below is best-effort. + tracing::warn!( + path = %crash_insights_path.display(), + "failed to persist worker exit record; will retry on next maintenance flush" + ); + } + // Delivery to hosted consumers must not silently drop the + // terminal event on backpressure: a full channel is retried via + // the bounded backlog (drained every tick) rather than dropped + // outright, and a closed channel is at least made observable via + // an error-level log and a counter. Either way the durable + // crash-insights record saved just above starts (and, on + // anything short of a successful handoff, stays) `Pending`, so + // it is always the authoritative record of what still needs + // delivering — including across a broker restart, which replays + // every still-`Pending` record. See `enqueue_hosted_agent_exit_event_tracked`. + super::event_loop::enqueue_hosted_agent_exit_event_tracked( + hosted_agent_event_tx, + hosted_agent_exit_backlog, + hosted_agent_exit_dropped_total, + Some(hosted_agent_exit_in_flight), + super::event_loop::hosted_agent_event_from_crash_record(&crash_record), + ); telemetry.track(TelemetryEvent::AgentCrash { cli: String::new(), @@ -416,6 +584,15 @@ impl BrokerRuntime { "signal": signal, "restart_count": restart_count, "delay_ms": delay.as_millis() as u64, + "reason": durable_reason, + "generation": generation_id, + "workspace_id": workspace_id, + "spawn_invocation_id": spawn_invocation_id, + "fleet_node_name": fleet_node_name, + "became_ready": ready_at.is_some(), + "spawned_at": spawned_at_unix, + "ready_at": ready_at_unix, + "exited_at": exited_at, }), ) .await; @@ -463,7 +640,20 @@ impl BrokerRuntime { agent_result_tokens.retain(|_, agent| agent != name); let _ = send_event( sdk_out_tx, - json!({"kind":"agent_permanently_dead","name":name,"reason":reason}), + json!({ + "kind":"agent_permanently_dead", + "name":name, + "reason":reason, + "exit_reason":durable_reason, + "generation":generation_id, + "workspace_id":workspace_id, + "spawn_invocation_id":spawn_invocation_id, + "fleet_node_name":fleet_node_name, + "became_ready":ready_at.is_some(), + "spawned_at":spawned_at_unix, + "ready_at":ready_at_unix, + "exited_at":exited_at, + }), ) .await; publish_agent_state_transition( @@ -537,7 +727,14 @@ impl BrokerRuntime { "code":code, "signal":signal, "reason": lifecycle_reason, - "generation": generation, + "generation": generation_id, + "workspace_id": workspace_id, + "spawn_invocation_id": spawn_invocation_id, + "fleet_node_name": fleet_node_name, + "became_ready": ready_at.is_some(), + "spawned_at": spawned_at_unix, + "ready_at": ready_at_unix, + "exited_at": exited_at, }), ) .await; diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index ef574c2f46..d2843380c2 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -856,6 +856,12 @@ pub(super) async fn spawn_worker_from_request( } } if let Some((token, invocation_id, session_ref)) = fleet_registration.take() { + // Carry correlation on this specific generation's handle, not + // just the by-name `fleet_inventory` entry: a same-name + // replacement can overwrite that entry before this + // generation is reaped, which would misattribute this + // invocation id to the wrong exit. See maintenance.rs reap. + workers.set_invocation_id(&name, invocation_id.clone()); super::fleet::record_fleet_inventory_agent( fleet_control_tx, fleet_inventory, @@ -1035,6 +1041,7 @@ mod tests { context_budget_pct: None, state: crate::worker::AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 15eaffe6f0..5f3ebe2335 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -124,6 +124,7 @@ async fn make_worker_registry_with_worker(name: &str) -> WorkerRegistry { context_budget_pct: None, state: AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); registry @@ -185,6 +186,7 @@ async fn make_worker_registry_with_stalled_worker(name: &str) -> WorkerRegistry context_budget_pct: None, state: AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); registry @@ -621,6 +623,12 @@ fn worker_event_runtime_fixture( ws_control_tx, relaycast_http, hosted_agent_event_tx, + hosted_agent_exit_backlog: std::collections::VecDeque::new(), + hosted_agent_exit_in_flight: std::collections::HashSet::new(), + hosted_agent_exit_dropped_total: 0, + hosted_delivery_result_rx: mpsc::channel(4).1, + hosted_delivery_result_open: true, + hosted_agent_exit_publish_failures_total: 0, pty_observability: HashMap::new(), api_rx, api_open: true, @@ -3006,6 +3014,14 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { .expect("present worker handle"); let _ = handle.child.start_kill(); let _ = handle.child.wait().await; + // The child is gone, but its old writer task can still win a race with + // the retry loop and accept one final pipe write on macOS. Replace the + // transport with a closed queue so every retry observes the intended + // writer failure deterministically while the worker remains present in + // the registry. + let (closed_tx, closed_rx) = mpsc::channel(1); + drop(closed_rx); + handle.command_tx = closed_tx; } assert!( workers.has_worker(worker_name), @@ -3063,9 +3079,21 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { break; } Ok(DeliveryAttemptOutcome::Attempted { attempts, .. }) => { + // `attempts` increments by exactly one per call regardless + // of outcome and is only ever capped at + // `MAX_DELIVERY_RETRIES` on the *failure* path (where it + // tracks the same budget as `failed_attempts`). A spurious + // successful write on a dead recipient (see the platform + // note above) resets `failed_attempts` and is deliberately + // *not* capped here — it is not a failure retry, so it must + // not be constrained by the failure budget (see + // `wait_delivery_successful_handoffs_do_not_exhaust_failure_budget`). + // The only invariant that must hold regardless of the + // success/failure mix is that attempts can't outrun the + // number of calls actually made. assert!( - attempts <= MAX_DELIVERY_RETRIES, - "retry attempts must stay within the retry cap" + attempts <= retry_index, + "attempts must never exceed the number of retry calls made so far" ); assert!( retry_index <= MAX_DELIVERY_RETRIES, @@ -6347,3 +6375,697 @@ async fn owned_cleanup_retries_delete_without_repeating_acknowledged_deregistrat .contains_key(&name)); fixture.runtime.workers.release("unrelated").await.unwrap(); } + +/// Regression for #1603 P2 #1: a same-name replacement's Fleet +/// re-registration overwrote the by-name `fleet_inventory` entry *before* +/// the old (already-dead) generation was reaped, so the maintenance tick's +/// by-name correlation lookup misattributed the *new* generation's +/// invocation id to the *old* generation's exit. Correlation must instead +/// travel on the exited generation's own handle. +#[cfg(unix)] +#[tokio::test] +async fn maintenance_tick_attributes_old_generation_exit_to_old_invocation_id() { + let (tx, _rx) = mpsc::channel(16); + let mut registry = WorkerRegistry::new( + tx, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests-reap-correlation"), + Instant::now(), + ); + let name = WorkerName::from("same-name-replacement"); + let old_generation = Uuid::new_v4(); + let old_child = tokio::process::Command::new("sh") + .args(["-c", "exit 0"]) + .spawn() + .expect("old-generation worker should spawn"); + let (old_command_tx, _old_command_rx) = mpsc::channel(16); + registry.workers.insert( + name.clone(), + WorkerHandle { + generation: old_generation, + spec: AgentSpec { + name: name.clone(), + runtime: AgentRuntime::Headless, + provider: None, + cli: None, + session_id: None, + harness_config: None, + model: None, + cwd: None, + team: None, + shadow_of: None, + shadow_mode: None, + args: Vec::new(), + channels: Vec::new(), + restart_policy: None, + }, + parent: None, + workspace_id: None, + child: old_child, + command_tx: old_command_tx, + harness_pid: None, + spawned_at: Instant::now(), + ready_at: Some(Instant::now()), + last_activity_at: Instant::now(), + context_budget_pct: None, + state: AgentWorkState::Working, + exit_reason: None, + // Set at spawn time (see `set_invocation_id` call sites) — this + // is the correlation for *this* generation specifically. + invocation_id: Some("inv-old-generation".to_string()), + }, + ); + // Let the old generation's process actually exit before the tick runs. + tokio::time::sleep(Duration::from_millis(50)).await; + + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + // Simulate the same-name replacement's Fleet registration landing in the + // by-name inventory map *before* maintenance reaps the old generation — + // this is the exact overwrite the bug report describes. + fixture.runtime.fleet_inventory.insert( + name.clone(), + crate::fleet_wire::InventoryAgent { + agent_id: "agent-new-generation".to_string(), + name: name.to_string(), + invocation_id: Some("inv-new-generation".to_string()), + session_ref: None, + }, + ); + + fixture.runtime.handle_maintenance_tick().await; + + let mut saw_exit_event = false; + while let Ok(envelope) = fixture._sdk_out_rx.try_recv() { + let payload = &envelope.payload; + if payload.get("kind").and_then(Value::as_str) == Some("agent_exited") + && payload.get("name").and_then(Value::as_str) == Some(name.as_str()) + { + assert_eq!( + payload.get("spawn_invocation_id").and_then(Value::as_str), + Some("inv-old-generation"), + "the old generation's exit must never be attributed to the \ + same-name replacement's invocation id: {payload:?}" + ); + saw_exit_event = true; + } + } + assert!(saw_exit_event, "expected an agent_exited sdk event"); + + // The durable crash-insights record must carry the same correlation. + let crash_json = fixture.runtime.crash_insights.to_json(); + let records = crash_json + .get("recent") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let record = records + .iter() + .find(|record| record.get("agent_name").and_then(Value::as_str) == Some(name.as_str())) + .expect("crash insight record for the old generation must exist"); + assert_eq!( + record.get("spawn_invocation_id").and_then(Value::as_str), + Some("inv-old-generation") + ); +} + +/// Regression for #1603 P2 #2: a full or closed `hosted_agent_event_tx` +/// channel must not silently drop the terminal `agent_exited` hosted event. +/// A full channel is retried via the bounded backlog on the next +/// maintenance tick; a closed channel cannot be retried but is at least +/// counted so the miss is observable. +#[tokio::test] +async fn hosted_agent_exit_event_survives_full_channel_and_reports_closed_channel() { + use super::event_loop::{enqueue_hosted_agent_exit_event_tracked, HostedAgentEvent}; + + // Full channel: the event must land in the backlog, not be dropped, and + // must be delivered once capacity frees up. + let (tx, mut rx) = mpsc::channel::(1); + // Occupy the only slot so the next enqueue observes `Full`. + tx.try_send(HostedAgentEvent { + name: "occupant".to_string(), + event_type: "agent_exited".to_string(), + payload: serde_json::Map::new(), + workspace_id: None, + dedupe_key: "occupant::gen-occupant".to_string(), + }) + .unwrap(); + + let mut backlog = std::collections::VecDeque::new(); + let mut dropped_total = 0u64; + enqueue_hosted_agent_exit_event_tracked( + &tx, + &mut backlog, + &mut dropped_total, + None, + HostedAgentEvent { + name: "full-channel-victim".to_string(), + event_type: "agent_exited".to_string(), + payload: serde_json::Map::new(), + workspace_id: None, + dedupe_key: "full-channel-victim::gen-full".to_string(), + }, + ); + assert_eq!( + backlog.len(), + 1, + "event must be held for retry, not dropped" + ); + assert_eq!(dropped_total, 0); + + // Drain the occupant so capacity frees up, then simulate the next tick's + // backlog drain — the held event must now be delivered. + let occupant = rx.recv().await.expect("occupant should be received"); + assert_eq!(occupant.name, "occupant"); + super::event_loop::drain_hosted_agent_exit_backlog_tracked( + &tx, + &mut backlog, + &mut dropped_total, + None, + ); + assert!(backlog.is_empty(), "backlog must drain once capacity frees"); + let delivered = rx + .recv() + .await + .expect("backlogged event must eventually be delivered"); + assert_eq!(delivered.name, "full-channel-victim"); + assert_eq!(dropped_total, 0); + + // Closed channel: cannot ever be retried, so it must not accumulate in + // the backlog — but the miss must be observable via the drop counter. + let (tx, rx) = mpsc::channel::(4); + drop(rx); + enqueue_hosted_agent_exit_event_tracked( + &tx, + &mut backlog, + &mut dropped_total, + None, + HostedAgentEvent { + name: "closed-channel-victim".to_string(), + event_type: "agent_exited".to_string(), + payload: serde_json::Map::new(), + workspace_id: None, + dedupe_key: "closed-channel-victim::gen-closed".to_string(), + }, + ); + assert!( + backlog.is_empty(), + "a closed channel can never succeed; must not be backlogged" + ); + assert_eq!( + dropped_total, 1, + "closed-channel miss must be observable via the drop counter" + ); +} + +/// Regression for #1603 P2 #3: the in-memory hosted publish queue must be +/// continuously replenished from durable `Pending` records after overflow +/// drops above `HOSTED_AGENT_EXIT_BACKLOG_CAP` (256), *within the running +/// process* — no broker restart required — and must never produce a +/// duplicate in-flight delivery for the same logical exit. +/// +/// Deterministic by construction: this drives +/// `replenish_hosted_agent_exit_backlog` directly, in a loop, simulating +/// each backlogged event's successful delivery outcome (exactly what +/// `BrokerRuntime::handle_hosted_delivery_outcome` does) between passes. +/// No sleeping, no real clock/publisher task — just repeated direct calls +/// until the durable pending set is fully drained. +#[test] +fn replenish_backlog_drains_a_large_pending_backlog_without_restart_or_duplicates() { + use super::event_loop::{replenish_hosted_agent_exit_backlog, HostedAgentEvent}; + use std::collections::{HashSet, VecDeque}; + + const TOTAL_PENDING: usize = 600; + const _: () = assert!( + TOTAL_PENDING > 500 && TOTAL_PENDING > super::event_loop::HOSTED_AGENT_EXIT_BACKLOG_CAP, + "fixture must exceed both the required minimums" + ); + + let mut crash_insights = crate::crash_insights::CrashInsights::new(); + for i in 0..TOTAL_PENDING { + let (category, description) = crate::crash_insights::CrashInsights::analyze(Some(1), None); + crash_insights.record(crate::crash_insights::CrashRecord { + agent_name: format!("worker-{i}"), + exit_code: Some(1), + signal: None, + timestamp: 1, + uptime_secs: 1, + category, + description, + workspace_id: None, + spawn_invocation_id: None, + generation: format!("gen-{i}"), + became_ready: true, + spawned_at: 0, + ready_at: None, + exited_at: 1, + exit_reason: None, + fleet_node_name: None, + hosted_delivery: crate::crash_insights::HostedDeliveryState::Pending, + }); + } + assert_eq!( + crash_insights.pending_hosted_deliveries().len(), + TOTAL_PENDING + ); + + let mut backlog: VecDeque = VecDeque::new(); + let mut in_flight: HashSet = HashSet::new(); + let mut ever_delivered: HashSet = HashSet::new(); + let mut passes = 0usize; + + // Bounded loop: draining `TOTAL_PENDING` records at up to + // `HOSTED_AGENT_EXIT_BACKLOG_CAP` per pass can never take more passes + // than that ratio (rounded up), plus a small safety margin — if it ever + // needed more, that itself would be a bug (a stall), so the test must + // fail loudly rather than hang. + let max_passes = TOTAL_PENDING.div_ceil(super::event_loop::HOSTED_AGENT_EXIT_BACKLOG_CAP) + 2; + + while !crash_insights.pending_hosted_deliveries().is_empty() { + passes += 1; + assert!( + passes <= max_passes, + "replenishment did not converge within the expected number of passes — \ + possible stall in the drain/replenish loop" + ); + + let replenished = + replenish_hosted_agent_exit_backlog(&crash_insights, &mut backlog, &mut in_flight); + assert!( + backlog.len() <= super::event_loop::HOSTED_AGENT_EXIT_BACKLOG_CAP, + "replenishment must never push the in-memory backlog past its cap" + ); + if replenished == 0 && backlog.is_empty() { + panic!( + "no progress possible: no records replenished and nothing in-flight to \ + resolve — pending records would be stuck forever" + ); + } + + // Simulate the publisher delivering and BrokerRuntime acking every + // currently-backlogged event successfully — exactly the + // `handle_hosted_delivery_outcome(success = true)` path — before + // the next replenishment pass. This is also the duplicate-delivery + // guard check: no dedupe key must ever be handed off a second time + // while it could still be in flight from a prior pass. + while let Some(event) = backlog.pop_front() { + assert!( + ever_delivered.insert(event.dedupe_key.clone()), + "duplicate in-flight delivery detected: '{}' was handed off more than once", + event.dedupe_key + ); + assert!( + crash_insights.mark_hosted_delivered(&event.dedupe_key), + "delivered dedupe key must correspond to a still-pending durable record" + ); + in_flight.remove(&event.dedupe_key); + } + } + + assert_eq!( + ever_delivered.len(), + TOTAL_PENDING, + "every pending record must eventually be delivered exactly once — none lost, none duplicated" + ); + assert_eq!( + crash_insights.pending_hosted_deliveries().len(), + 0, + "no pending exit may remain after the drain completes" + ); + assert!( + in_flight.is_empty(), + "in-flight tracking must be fully cleared once every delivery is acknowledged" + ); + assert!( + passes > 1, + "fixture must actually exercise multiple replenishment passes (backlog cap < total pending)" + ); +} + +/// Once a durable crash record's dedupe key has been acknowledged (marked +/// `Delivered`), replaying the outbox (as a restart would) must not +/// reconstruct or redeliver it. +#[tokio::test] +async fn delivered_dedupe_key_is_excluded_from_replay() { + use super::event_loop::{ + enqueue_hosted_agent_exit_event_tracked, hosted_agent_event_from_crash_record, + reload_pending_hosted_agent_exit_backlog, HostedAgentEvent, + }; + use crate::crash_insights::{CrashInsights, CrashRecord, HostedDeliveryState}; + + let dir = tempfile::tempdir().unwrap(); + let crash_insights_path = dir.path().join("crashes.json"); + let mut crash_insights = CrashInsights::new(); + + let record = CrashRecord { + agent_name: "delivered-agent".to_string(), + exit_code: Some(1), + signal: None, + timestamp: 1, + uptime_secs: 1, + category: crate::crash_insights::CrashCategory::Error, + description: "Exited with code 1".to_string(), + workspace_id: None, + spawn_invocation_id: None, + generation: "gen-delivered".to_string(), + became_ready: true, + spawned_at: 0, + ready_at: None, + exited_at: 1, + exit_reason: None, + fleet_node_name: None, + hosted_delivery: HostedDeliveryState::Pending, + }; + crash_insights.record(record.clone()); + crash_insights.save(&crash_insights_path).unwrap(); + + // Room for one, so the handoff to the publisher's channel succeeds + // immediately — but a handoff is not delivery: the critical fix for + // #1750's review is that this must NOT mark the durable record + // delivered by itself. Only the publisher's real HTTP-success outcome + // (simulated here via `mark_hosted_delivered_and_persist`, the same + // call `BrokerRuntime::handle_hosted_delivery_outcome` makes) may do + // that. + let (tx, mut rx) = mpsc::channel::(1); + let mut backlog = std::collections::VecDeque::new(); + let mut dropped_total = 0u64; + enqueue_hosted_agent_exit_event_tracked( + &tx, + &mut backlog, + &mut dropped_total, + None, + hosted_agent_event_from_crash_record(&record), + ); + let handed_off = rx.recv().await.expect("event should reach the publisher"); + assert_eq!(handed_off.name, "delivered-agent"); + assert_eq!( + crash_insights.pending_hosted_deliveries().len(), + 1, + "a mere channel handoff must not mark the durable record delivered — \ + only a confirmed Relaycast HTTP success may (critical fix for #1750 review)" + ); + + // Restart replay must still see this as pending, since the publisher + // has not yet confirmed Relaycast accepted it. + let reloaded_before_confirmation = CrashInsights::load(&crash_insights_path); + assert_eq!( + reloaded_before_confirmation + .pending_hosted_deliveries() + .len(), + 1, + "on-disk record must remain pending until real delivery is confirmed" + ); + + // Now simulate the publisher confirming the real HTTP call succeeded — + // this is the only path that may flip the durable record to Delivered. + super::event_loop::mark_hosted_delivered_and_persist( + &mut crash_insights, + &crash_insights_path, + true, + &record.dedupe_key(), + ); + assert_eq!( + crash_insights.pending_hosted_deliveries().len(), + 0, + "confirmed HTTP success must mark the durable record delivered" + ); + + let reloaded = CrashInsights::load(&crash_insights_path); + assert_eq!( + reloaded.pending_hosted_deliveries().len(), + 0, + "delivered state must have been persisted, not just held in memory" + ); + + let mut replay_dropped_total = 0u64; + let replayed = reload_pending_hosted_agent_exit_backlog(&reloaded, &mut replay_dropped_total); + assert!( + replayed.is_empty(), + "a delivered record must not be replayed on restart" + ); +} + +/// Regression for #1603 P2 #2, exercised at the full maintenance-tick level: +/// a hosted event channel with zero remaining capacity at reap time must not +/// silently lose the terminal `agent_exited` event — it is retried from the +/// bounded backlog on the very next tick once the channel has capacity. +#[cfg(unix)] +#[tokio::test] +async fn maintenance_tick_backlogs_and_redelivers_agent_exited_when_hosted_channel_is_full() { + let (tx, _rx) = mpsc::channel(16); + let mut registry = WorkerRegistry::new( + tx, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests-hosted-backlog"), + Instant::now(), + ); + let name = WorkerName::from("hosted-backlog-victim"); + let generation = Uuid::new_v4(); + let child = tokio::process::Command::new("sh") + .args(["-c", "exit 0"]) + .spawn() + .expect("worker should spawn"); + let (command_tx, _command_rx) = mpsc::channel(16); + registry.workers.insert( + name.clone(), + WorkerHandle { + generation, + spec: AgentSpec { + name: name.clone(), + runtime: AgentRuntime::Headless, + provider: None, + cli: None, + session_id: None, + harness_config: None, + model: None, + cwd: None, + team: None, + shadow_of: None, + shadow_mode: None, + args: Vec::new(), + channels: Vec::new(), + restart_policy: None, + }, + parent: None, + workspace_id: None, + child, + command_tx, + harness_pid: None, + spawned_at: Instant::now(), + ready_at: Some(Instant::now()), + last_activity_at: Instant::now(), + context_budget_pct: None, + state: AgentWorkState::Working, + exit_reason: None, + invocation_id: Some("inv-hosted-backlog".to_string()), + }, + ); + tokio::time::sleep(Duration::from_millis(50)).await; + + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + // Replace the fixture's hosted-agent-event channel with a zero-capacity + // one that is already full, forcing `try_send` to observe `Full` when + // the tick emits `agent_exited`. + let (hosted_tx, mut hosted_rx) = mpsc::channel(1); + hosted_tx + .try_send(super::event_loop::HostedAgentEvent { + name: "occupant".to_string(), + event_type: "agent_exited".to_string(), + payload: serde_json::Map::new(), + workspace_id: None, + dedupe_key: "occupant::gen-occupant".to_string(), + }) + .unwrap(); + fixture.runtime.hosted_agent_event_tx = hosted_tx; + + fixture.runtime.handle_maintenance_tick().await; + + // The channel was full, so the real terminal event must be sitting in + // the backlog now, not lost. + assert_eq!(fixture.runtime.hosted_agent_exit_backlog.len(), 1); + assert_eq!(fixture.runtime.hosted_agent_exit_dropped_total, 0); + + // Drain the occupant to free capacity, then run another tick (a no-op + // for this already-reaped worker otherwise) — the backlog drain at the + // top of the tick must redeliver the held event. + let occupant = hosted_rx.recv().await.expect("occupant received"); + assert_eq!(occupant.name, "occupant"); + fixture.runtime.handle_maintenance_tick().await; + assert!(fixture.runtime.hosted_agent_exit_backlog.is_empty()); + let redelivered = hosted_rx + .recv() + .await + .expect("backlogged agent_exited must be redelivered"); + assert_eq!(redelivered.name, name.as_str()); + assert_eq!(redelivered.event_type, "agent_exited"); + assert_eq!(fixture.runtime.hosted_agent_exit_dropped_total, 0); +} + +/// End-to-end durable-outbox regression: a worker exits while the hosted +/// event channel is *closed* (the unrecoverable-for-this-process-lifetime +/// case), so the in-memory backlog can never redeliver it — the only way it +/// can ever reach a hosted consumer is a full broker restart replaying the +/// still-`Pending` durable crash record from disk. This simulates exactly +/// that: build a fresh `CrashInsights` from the same on-disk path a restart +/// would load, replay its pending entries into a new backlog, wire that +/// backlog to a live (non-closed) channel, and confirm the very next +/// maintenance tick's backlog-drain step redelivers the event that the +/// prior "process" could never have delivered itself. +#[cfg(unix)] +#[tokio::test] +async fn restart_before_drain_replays_pending_delivery_from_disk() { + let (tx, _rx) = mpsc::channel(16); + let mut registry = WorkerRegistry::new( + tx, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests-restart-replay"), + Instant::now(), + ); + let name = WorkerName::from("restart-replay-victim"); + let generation = Uuid::new_v4(); + let child = tokio::process::Command::new("sh") + .args(["-c", "exit 0"]) + .spawn() + .expect("worker should spawn"); + let (command_tx, _command_rx) = mpsc::channel(16); + registry.workers.insert( + name.clone(), + WorkerHandle { + generation, + spec: AgentSpec { + name: name.clone(), + runtime: AgentRuntime::Headless, + provider: None, + cli: None, + session_id: None, + harness_config: None, + model: None, + cwd: None, + team: None, + shadow_of: None, + shadow_mode: None, + args: Vec::new(), + channels: Vec::new(), + restart_policy: None, + }, + parent: None, + workspace_id: None, + child, + command_tx, + harness_pid: None, + spawned_at: Instant::now(), + ready_at: Some(Instant::now()), + last_activity_at: Instant::now(), + context_budget_pct: None, + state: AgentWorkState::Working, + exit_reason: None, + invocation_id: Some("inv-restart-replay".to_string()), + }, + ); + tokio::time::sleep(Duration::from_millis(50)).await; + + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + // Persistence must actually be on disk for a "restart" to have anything + // to reload. + fixture.runtime.paths.persist = true; + let crash_insights_path = fixture.runtime.crash_insights_path.clone(); + + // Close the hosted channel entirely — the unrecoverable-within-this- + // process case. The durable record is the only thing that can save this + // delivery now. + let (hosted_tx, hosted_rx) = mpsc::channel(4); + drop(hosted_rx); + fixture.runtime.hosted_agent_event_tx = hosted_tx; + + fixture.runtime.handle_maintenance_tick().await; + + // Closed channel: never backlogged in-memory, and counted as a miss — + // but the durable record on disk must still be `Pending`. + assert!(fixture.runtime.hosted_agent_exit_backlog.is_empty()); + assert_eq!(fixture.runtime.hosted_agent_exit_dropped_total, 1); + assert_eq!( + fixture + .runtime + .crash_insights + .pending_hosted_deliveries() + .len(), + 1 + ); + + // Simulate the restart: load a brand-new `CrashInsights` from the same + // path (nothing shared with the "crashed" process above) and replay its + // pending entries into a fresh backlog. + let reloaded = crate::crash_insights::CrashInsights::load(&crash_insights_path); + assert_eq!(reloaded.pending_hosted_deliveries().len(), 1); + let mut replay_dropped_total = 0u64; + let replayed_backlog = super::event_loop::reload_pending_hosted_agent_exit_backlog( + &reloaded, + &mut replay_dropped_total, + ); + assert_eq!(replayed_backlog.len(), 1); + assert_eq!(replayed_backlog[0].name, name.as_str()); + assert_eq!(replay_dropped_total, 0); + + // Wire the replayed backlog into a fresh runtime with a live channel — + // the very next maintenance tick's top-of-tick drain must redeliver it. + let empty_registry = WorkerRegistry::new( + mpsc::channel(16).0, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests-restart-replay-2"), + Instant::now(), + ); + let mut restarted = worker_event_runtime_fixture(empty_registry, HashMap::new()); + restarted.runtime.crash_insights = reloaded; + restarted.runtime.crash_insights_path = crash_insights_path; + // Mirror what `init.rs` does for a real restart: seed the in-flight + // dedupe guard with every record replayed into the backlog, so the + // maintenance tick's replenishment pass doesn't immediately re-enqueue + // a second in-memory copy of the same still-`Pending` record. + restarted.runtime.hosted_agent_exit_in_flight = replayed_backlog + .iter() + .map(|event| event.dedupe_key.clone()) + .collect(); + restarted.runtime.hosted_agent_exit_backlog = replayed_backlog; + let (live_tx, mut live_rx) = mpsc::channel(4); + restarted.runtime.hosted_agent_event_tx = live_tx; + + restarted.runtime.handle_maintenance_tick().await; + + assert!(restarted.runtime.hosted_agent_exit_backlog.is_empty()); + let redelivered = live_rx + .recv() + .await + .expect("restart replay must redeliver the event the crashed process never could"); + assert_eq!(redelivered.name, name.as_str()); + assert_eq!(redelivered.event_type, "agent_exited"); + + // Critical fix for #1750's review: reaching the publisher's queue again + // (a mere handoff) must NOT yet mark the durable record delivered — it + // stays `Pending` until the publisher's real Relaycast HTTP call is + // confirmed. Only then (simulated here exactly as + // `BrokerRuntime::handle_hosted_delivery_outcome` would on a real + // success outcome) does it flip to `Delivered`. + assert_eq!( + restarted + .runtime + .crash_insights + .pending_hosted_deliveries() + .len(), + 1, + "replay handoff alone must not mark the record delivered before real hosted success" + ); + super::event_loop::mark_hosted_delivered_and_persist( + &mut restarted.runtime.crash_insights, + &restarted.runtime.crash_insights_path, + true, + &redelivered.dedupe_key, + ); + assert_eq!( + restarted + .runtime + .crash_insights + .pending_hosted_deliveries() + .len(), + 0, + "redelivery must mark the durable record delivered" + ); +} diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 8cecd80c27..4e2ec3eb3a 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -155,6 +155,10 @@ fn enqueue_pty_event( event_type: event_type.to_string(), payload, workspace_id: state.workspace_id.clone(), + // Best-effort observability events (non-terminal) never go through + // the durable agent_exited outbox/backlog, so this key is only ever + // used for logging context, never for dedup lookups. + dedupe_key: format!("{}::pty::{}", name, state.sequence), }) { tracing::warn!(worker = %name, error = %error, "Relaycast PTY observability queue is full or closed"); } @@ -594,11 +598,21 @@ fn hosted_agent_event( if let Some(timestamp) = payload.get("timestamp") { event_payload.insert("timestamp".to_string(), timestamp.clone()); } + let dedupe_key = format!( + "{}::relayed::{}::{}", + name, + event_type, + payload.get("sequence").cloned().unwrap_or(Value::Null) + ); Some(HostedAgentEvent { name: name.to_string(), event_type, payload: event_payload, workspace_id, + // Relayed worker-protocol events (not the broker-synthesized + // terminal `agent_exited`) never go through the durable outbox + // either; this key is only for logging/uniqueness, not dedup. + dedupe_key, }) } diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index 885c271209..f961bd71c4 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -193,14 +193,28 @@ pub(crate) struct WorkerHandle { pub(crate) context_budget_pct: Option, pub(crate) state: AgentWorkState, pub(crate) exit_reason: Option, + /// Fleet correlation for this specific spawned generation. Captured at + /// spawn time (or immediately after, once the fleet registration + /// resolves) and carried on the handle itself rather than looked up by + /// worker name at reap time — a same-name replacement worker can + /// overwrite a by-name `fleet_inventory` entry before the old + /// generation is reaped, which would otherwise misattribute the new + /// generation's invocation id to the old generation's exit. + pub(crate) invocation_id: Option, } +/// `invocation_id` is the last element so this stays append-only for +/// existing positional destructuring call sites. pub(crate) type ExitedWorker = ( WorkerName, Uuid, Option, Option, Option, + Option, + Instant, + Option, + Option, ); #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -1312,6 +1326,7 @@ impl WorkerRegistry { context_budget_pct: None, state: AgentWorkState::Working, exit_reason: None, + invocation_id: None, }; self.workers.insert(spec.name.clone(), handle); @@ -1564,6 +1579,16 @@ impl WorkerRegistry { Ok(()) } + /// Record the Fleet invocation id that correlates to the currently live + /// generation of `name`, once it is known (fleet registration resolves + /// asynchronously after the process is spawned). A no-op if the worker + /// is no longer registered under that name. + pub(crate) fn set_invocation_id(&mut self, name: &WorkerName, invocation_id: Option) { + if let Some(handle) = self.workers.get_mut(name) { + handle.invocation_id = invocation_id; + } + } + pub(crate) async fn reap_exited(&mut self) -> Result> { let names: Vec = self.workers.keys().cloned().collect(); let mut exited = Vec::new(); @@ -1625,16 +1650,23 @@ impl WorkerRegistry { None }; if let Some(orphan) = orphaned { - let generation = self + let (generation, workspace_id, spawned_at, ready_at, reason, invocation_id) = self .workers .get(&name) - .expect("orphaned worker must still be registered") - .generation; - let reason = self - .workers - .get(&name) - .and_then(|handle| handle.exit_reason.clone()) - .or_else(|| Some(orphan.reason().to_string())); + .map(|handle| { + ( + handle.generation, + handle.workspace_id.clone(), + handle.spawned_at, + handle.ready_at, + handle + .exit_reason + .clone() + .or_else(|| Some(orphan.reason().to_string())), + handle.invocation_id.clone(), + ) + }) + .expect("orphaned worker must still be registered"); if let Some(handle) = self.workers.get_mut(&name) { tracing::warn!( worker = %name, @@ -1667,15 +1699,34 @@ impl WorkerRegistry { } self.workers.remove(&name); self.initial_tasks.remove(&name); - exited.push((name, generation, None, None, reason)); + exited.push(( + name, + generation, + None, + None, + reason, + workspace_id, + spawned_at, + ready_at, + invocation_id, + )); continue; } if let Some(status) = status { - let generation = self + let (generation, workspace_id, spawned_at, ready_at, reason, invocation_id) = self .workers .get(&name) - .expect("exited worker must still be registered") - .generation; + .map(|handle| { + ( + handle.generation, + handle.workspace_id.clone(), + handle.spawned_at, + handle.ready_at, + handle.exit_reason.clone(), + handle.invocation_id.clone(), + ) + }) + .expect("exited worker must still be registered"); let code = status.code(); #[cfg(unix)] let signal = { @@ -1684,26 +1735,47 @@ impl WorkerRegistry { }; #[cfg(not(unix))] let signal: Option = None; - let reason = self - .workers - .get(&name) - .and_then(|handle| handle.exit_reason.clone()); self.workers.remove(&name); self.initial_tasks.remove(&name); - exited.push((name, generation, code, signal, reason)); + exited.push(( + name, + generation, + code, + signal, + reason, + workspace_id, + spawned_at, + ready_at, + invocation_id, + )); } else if gone_via_kill0 { - let generation = self - .workers - .get(&name) - .expect("gone worker must still be registered") - .generation; - let reason = self + let (generation, workspace_id, spawned_at, ready_at, reason, invocation_id) = self .workers .get(&name) - .and_then(|handle| handle.exit_reason.clone()); + .map(|handle| { + ( + handle.generation, + handle.workspace_id.clone(), + handle.spawned_at, + handle.ready_at, + handle.exit_reason.clone(), + handle.invocation_id.clone(), + ) + }) + .expect("gone worker must still be registered"); self.workers.remove(&name); self.initial_tasks.remove(&name); - exited.push((name, generation, None, None, reason)); + exited.push(( + name, + generation, + None, + None, + reason, + workspace_id, + spawned_at, + ready_at, + invocation_id, + )); } } Ok(exited) @@ -2874,6 +2946,170 @@ sleep 30 kill(Pid::from_raw(pid as i32), None).is_ok() } + #[cfg(unix)] + #[tokio::test] + async fn reap_exited_preserves_worker_correlation_metadata() { + let mut reg = make_registry(vec![]); + let name = WorkerName::from("correlated-exit"); + let generation = Uuid::new_v4(); + let workspace_id = crate::ids::WorkspaceId::new("workspace-1"); + let spawned_at = Instant::now() - Duration::from_secs(5); + let ready_at = Instant::now() - Duration::from_secs(2); + let child = Command::new("sh") + .args(["-c", "exit 7"]) + .spawn() + .expect("short-lived worker should spawn"); + let (command_tx, _command_rx) = mpsc::channel(WORKER_WRITE_QUEUE_CAPACITY); + reg.workers.insert( + name.clone(), + WorkerHandle { + generation, + spec: spec_for_test(name.as_str()), + parent: None, + workspace_id: Some(workspace_id.clone()), + child, + command_tx, + harness_pid: None, + spawned_at, + ready_at: Some(ready_at), + last_activity_at: ready_at, + context_budget_pct: None, + state: AgentWorkState::Working, + exit_reason: Some("worker_write_failed".to_string()), + invocation_id: Some("inv-original".to_string()), + }, + ); + + tokio::time::sleep(Duration::from_millis(50)).await; + let exited = reg.reap_exited().await.expect("reap should succeed"); + assert_eq!(exited.len(), 1); + let ( + exited_name, + exited_generation, + exit_code, + _signal, + exit_reason, + exited_workspace, + exited_spawned_at, + exited_ready_at, + exited_invocation_id, + ) = &exited[0]; + assert_eq!(exited_name, &name); + assert_eq!(*exited_generation, generation); + assert_eq!(*exit_code, Some(7)); + assert_eq!(exit_reason.as_deref(), Some("worker_write_failed")); + assert_eq!(exited_workspace.as_ref(), Some(&workspace_id)); + assert_eq!(*exited_spawned_at, spawned_at); + assert_eq!(*exited_ready_at, Some(ready_at)); + assert_eq!(exited_invocation_id.as_deref(), Some("inv-original")); + } + + /// Regression for the same-name-replacement correlation race: the exited + /// generation's Fleet invocation id must come from the handle captured + /// at reap time, not from a by-name lookup performed after a same-name + /// replacement has already overwritten that entry with its own id. + /// `reap_exited` itself has no by-name lookup (the correlation travels + /// on the handle), so this proves the two generations' invocation ids + /// can never cross even when the replacement is registered before the + /// old generation is reaped. + #[cfg(unix)] + #[tokio::test] + async fn reap_exited_does_not_cross_invocation_ids_across_same_name_replacement() { + let mut reg = make_registry(vec![]); + let name = WorkerName::from("replaced-worker"); + let old_generation = Uuid::new_v4(); + let old_child = Command::new("sh") + .args(["-c", "exit 3"]) + .spawn() + .expect("old-generation worker should spawn"); + let (old_command_tx, _old_command_rx) = mpsc::channel(WORKER_WRITE_QUEUE_CAPACITY); + reg.workers.insert( + name.clone(), + WorkerHandle { + generation: old_generation, + spec: spec_for_test(name.as_str()), + parent: None, + workspace_id: None, + child: old_child, + command_tx: old_command_tx, + harness_pid: None, + spawned_at: Instant::now(), + ready_at: Some(Instant::now()), + last_activity_at: Instant::now(), + context_budget_pct: None, + state: AgentWorkState::Working, + exit_reason: None, + invocation_id: Some("inv-old-generation".to_string()), + }, + ); + + // Let the old generation's process exit, but do not reap it yet — + // model a same-name respawn racing ahead of the maintenance reap + // sweep for the dead generation. + tokio::time::sleep(Duration::from_millis(50)).await; + + // A same-name replacement now overwrites the registry entry (a real + // spawn always removes the old entry first; mirror that here) and + // carries a *different* invocation id — exactly the scenario that + // corrupted correlation when the lookup was by name instead of + // generation. + let new_generation = Uuid::new_v4(); + let new_child = Command::new("sleep") + .arg("30") + .spawn() + .expect("new-generation worker should spawn"); + let new_pid = new_child.id().expect("new child has a pid"); + let (new_command_tx, _new_command_rx) = mpsc::channel(WORKER_WRITE_QUEUE_CAPACITY); + reg.workers.insert( + name.clone(), + WorkerHandle { + generation: new_generation, + spec: spec_for_test(name.as_str()), + parent: None, + workspace_id: None, + child: new_child, + command_tx: new_command_tx, + harness_pid: None, + spawned_at: Instant::now(), + ready_at: None, + last_activity_at: Instant::now(), + context_budget_pct: None, + state: AgentWorkState::Working, + exit_reason: None, + invocation_id: Some("inv-new-generation".to_string()), + }, + ); + + // Only the live (new) generation remains under this name; reaping + // now must report the new generation as still alive (no exit), never + // resurrecting the old generation's exit under the new id. + let exited = reg.reap_exited().await.expect("reap should succeed"); + assert!( + exited.is_empty(), + "the new generation is alive and must not be reaped: {exited:?}" + ); + let live = reg + .workers + .get(&name) + .expect("new generation remains registered"); + assert_eq!(live.generation, new_generation); + assert_eq!(live.invocation_id.as_deref(), Some("inv-new-generation")); + + // Clean up the still-running replacement process. + use nix::{ + sys::signal::{kill, Signal}, + unistd::Pid, + }; + let _ = kill(Pid::from_raw(new_pid as i32), Signal::SIGKILL); + let _ = reg + .workers + .get_mut(&name) + .expect("worker still registered") + .child + .wait() + .await; + } + #[cfg(unix)] #[tokio::test] async fn cleanup_rejected_spawn_terminates_a_still_alive_child_and_removes_it() { @@ -2922,6 +3158,7 @@ sleep 30 context_budget_pct: None, state: AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); @@ -2975,6 +3212,7 @@ sleep 30 context_budget_pct: None, state: AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); @@ -3036,6 +3274,7 @@ sleep 30 context_budget_pct: None, state: AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); @@ -3096,6 +3335,7 @@ sleep 30 context_budget_pct: None, state: AgentWorkState::Working, exit_reason: None, + invocation_id: None, }, ); diff --git a/crates/relay-pty/Cargo.toml b/crates/relay-pty/Cargo.toml index 061727a20d..189b3da26c 100644 --- a/crates/relay-pty/Cargo.toml +++ b/crates/relay-pty/Cargo.toml @@ -17,6 +17,7 @@ portable-pty = "0.8" regex = "1.11" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +tempfile = "3.19" thiserror = "2.0" tokio = { version = "1.44", features = ["sync", "time", "rt", "macros", "process", "io-util"] } tracing = "0.1" @@ -26,6 +27,3 @@ libc = "0.2" [target.'cfg(windows)'.dependencies] windows-sys = { version = "0.59", features = ["Win32_Foundation", "Win32_System_IO", "Win32_System_Threading"] } - -[dev-dependencies] -tempfile = "3.19" diff --git a/crates/relay-pty/src/crash_insights.rs b/crates/relay-pty/src/crash_insights.rs index c0ca696921..fb014eecbd 100644 --- a/crates/relay-pty/src/crash_insights.rs +++ b/crates/relay-pty/src/crash_insights.rs @@ -2,6 +2,29 @@ //! //! Classifies agent crashes by exit code and signal, maintains a bounded //! history, detects patterns, and computes a health score. +//! +//! ## Storage tradeoff: diagnostics vs. the durable pending outbox +//! +//! This module deliberately applies two different durability policies to +//! two different halves of the same on-disk file: +//! +//! - **Diagnostics** (patterns, health score, "recent" crash history for +//! already-delivered exits) are bounded and *lossy by design*: once +//! [`CrashInsights::record`] pushes the store past `max_records`, the +//! oldest already-delivered records are evicted to keep the file bounded. +//! Losing old, already-delivered diagnostic history is an acceptable +//! tradeoff — it is presentation/analysis data, not the mechanism backing +//! at-least-once delivery. +//! - **The durable pending outbox** — [`CrashRecord`]s whose +//! [`HostedDeliveryState`] is still `Pending` — backs at-least-once hosted +//! `agent_exited` delivery and is **never silently evicted** under +//! retention pressure, even if that means the on-disk file temporarily +//! grows past `max_records` while deliveries are outstanding. When +//! retention pressure hits a store that is entirely (or mostly) pending +//! records, `record` skips eviction, logs loudly, and increments +//! [`CrashInsights::retention_pressure_total`] — an explicit, +//! operator-visible counter (surfaced via [`CrashInsights::to_json`] and +//! the broker's `GetCrashInsights` API) rather than a silent drop. use std::collections::HashMap; use std::path::Path; @@ -24,6 +47,44 @@ pub enum CrashCategory { Unknown, } +/// Delivery state of a crash record's corresponding hosted `agent_exited` +/// event. +/// +/// This is the durable half of the hosted-delivery outbox: every exit is +/// recorded [`Pending`](HostedDeliveryState::Pending) in the same atomic +/// write as the rest of the crash record (see [`CrashInsights::record`] / +/// [`CrashInsights::save`]), *before* the broker attempts to hand the event +/// to the hosted publisher channel. A successful handoff into that +/// channel (a successful `mpsc::Sender::try_send`) is **not** an +/// acknowledgment — it only proves the publisher task received the event, +/// not that Relaycast did. The record flips to +/// [`Delivered`](HostedDeliveryState::Delivered) only after the publisher +/// task's own HTTP call to Relaycast actually succeeds and reports that +/// outcome back over its result channel to +/// `BrokerRuntime::handle_hosted_delivery_outcome`, and that transition is +/// itself persisted immediately. A broker crash at any point before that +/// confirmed-success report — including while the handoff, the HTTP call, +/// or its retries are still in flight — therefore always leaves the +/// on-disk record `Pending`; it never claims delivery happened when it +/// didn't. +/// +/// On restart, every still-`Pending` record is a candidate for replay (see +/// `hosted_agent_event_from_crash_record` in the broker crate), giving +/// at-least-once delivery: a record may occasionally be replayed after it +/// was in fact delivered (e.g. the success write raced a crash), which is +/// why replay is keyed by [`CrashRecord::dedupe_key`] so hosted consumers can +/// idempotently discard a duplicate. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum HostedDeliveryState { + /// Not yet handed off to the hosted publisher channel (or the handoff + /// failed and was not retried). Eligible for replay on restart. + #[default] + Pending, + /// Successfully handed to the hosted publisher channel at least once. + Delivered, +} + /// A single crash record. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct CrashRecord { @@ -34,6 +95,52 @@ pub struct CrashRecord { pub uptime_secs: u64, pub category: CrashCategory, pub description: String, + /// Broker workspace that owned this worker, when the spawn was workspace-scoped. + #[serde(default)] + pub workspace_id: Option, + /// Fleet action invocation that created this worker, when available. + #[serde(default)] + pub spawn_invocation_id: Option, + /// Process generation; same-name workers must remain independently queryable. + #[serde(default)] + pub generation: String, + /// Whether the broker observed `worker_ready` for this process generation. + #[serde(default)] + pub became_ready: bool, + /// Unix timestamp (seconds) at which the wrapper process was spawned. + #[serde(default)] + pub spawned_at: u64, + /// Unix timestamp (seconds) at which the worker reported ready. + #[serde(default)] + pub ready_at: Option, + /// Unix timestamp (seconds) at which the broker reaped this generation. + #[serde(default)] + pub exited_at: u64, + /// Bounded, broker-derived explanation for why the process was reaped. + #[serde(default)] + pub exit_reason: Option, + /// Fleet node name that hosted this generation. + #[serde(default)] + pub fleet_node_name: Option, + /// Durable delivery state of the corresponding hosted `agent_exited` + /// event. See [`HostedDeliveryState`]. Defaults to `Pending` so records + /// written by older brokers (before this field existed) are always + /// eligible for replay rather than silently treated as delivered. + #[serde(default)] + pub hosted_delivery: HostedDeliveryState, +} + +impl CrashRecord { + /// Stable idempotency key for this generation's hosted delivery: + /// `agent_name` plus `generation`. A same-name replacement worker gets a + /// new `generation` (see `WorkerHandle`), so this key never collides + /// across two different process lifetimes of the same agent name, and + /// is stable across broker restarts (unlike, say, a locally-assigned + /// sequence number) so hosted consumers can dedupe a replayed delivery + /// against one they already saw before a restart. + pub fn dedupe_key(&self) -> String { + format!("{}::{}", self.agent_name, self.generation) + } } /// A detected crash pattern (grouping). @@ -50,6 +157,35 @@ pub struct CrashInsights { records: Vec, #[serde(default = "default_max_records")] max_records: usize, + /// True when in-memory state has changed since the last confirmed + /// successful [`save`](CrashInsights::save). Set on every mutation and, + /// crucially, re-set on a *failed* [`persist`](CrashInsights::persist) + /// call so a later flush attempt is guaranteed rather than the failure + /// being silently swallowed after a single log line. Never persisted: + /// a freshly loaded snapshot is by definition not dirty relative to + /// itself. + #[serde(skip)] + dirty: bool, + /// Count of [`persist`](CrashInsights::persist) calls whose underlying + /// [`save`](CrashInsights::save) failed, since process start. Purely + /// observability — never persisted to disk — so a transient write + /// failure (full disk, permissions, etc.) is visible to operators via + /// the broker's status/API surface (`GetCrashInsights`), not only in + /// logs that can scroll away unnoticed. + #[serde(skip)] + save_failures_total: u64, + /// Count of times [`record`](CrashInsights::record) hit generic + /// retention pressure (more than `max_records` records) but could not + /// evict anything because every record above the cap was still a + /// pending hosted-delivery outbox entry. Never persisted. This is the + /// operator-visible signal for the documented pressure policy: the + /// durable pending outbox is *never* silently evicted, so under + /// sustained pressure the on-disk file grows past `max_records` instead + /// — this counter says exactly how often that happened, so an operator + /// (or alert) can see the pressure building rather than discovering an + /// unbounded file after the fact. + #[serde(skip)] + retention_pressure_total: u64, } fn default_max_records() -> usize { @@ -67,6 +203,9 @@ impl CrashInsights { Self { records: Vec::new(), max_records: 500, + dirty: false, + save_failures_total: 0, + retention_pressure_total: 0, } } @@ -110,10 +249,32 @@ impl CrashInsights { /// Record a crash. Trims oldest records if over the limit. pub fn record(&mut self, crash: CrashRecord) { + self.dirty = true; self.records.push(crash); if self.records.len() > self.max_records { - let excess = self.records.len() - self.max_records; - self.records.drain(..excess); + let mut excess = self.records.len() - self.max_records; + // Preserve pending hosted exits first: these records are the + // durable outbox source and must survive long outages even when + // generic crash retention is under pressure. + while excess > 0 { + if let Some(index) = self + .records + .iter() + .position(|record| record.hosted_delivery != HostedDeliveryState::Pending) + { + self.records.remove(index); + excess -= 1; + } else { + self.retention_pressure_total += 1; + tracing::error!( + max_records = self.max_records, + pending_hosted_deliveries = self.pending_hosted_deliveries().len(), + retention_pressure_total = self.retention_pressure_total, + "crash-insights retention is full of pending hosted exits; preserving durable outbox records above the generic cap" + ); + break; + } + } } } @@ -175,6 +336,121 @@ impl CrashInsights { self.records.len() } + /// Records whose hosted `agent_exited` delivery has not yet succeeded, + /// in the original (chronological) order they were recorded. + /// + /// This is the durable hosted-delivery outbox's replay source: the + /// broker walks this on startup. `max_records` (via + /// [`CrashInsights::record`]) bounds *diagnostic* history only — the + /// oldest already-`Delivered` records are evicted first to keep the + /// on-disk file from growing without limit. Still-`Pending` records are + /// never evicted by that cap: a broker that is offline (or whose hosted + /// channel stays closed) long enough to accumulate more than + /// `max_records` outstanding deliveries will let the on-disk file grow + /// past `max_records` instead of losing any of them. That pressure is + /// logged loudly (and counted via `retention_pressure_total`) whenever + /// it happens; it is never silent, and it is never lossy for pending + /// deliveries. + pub fn pending_hosted_deliveries(&self) -> Vec<&CrashRecord> { + self.records + .iter() + .filter(|record| record.hosted_delivery == HostedDeliveryState::Pending) + .collect() + } + + /// Mark the most recent record matching `dedupe_key` as delivered and + /// return whether a matching (still-pending) record was found. + /// + /// Callers should persist ([`CrashInsights::save`]) immediately after a + /// successful call so the transition is durable before the process could + /// crash again. Searches from the end since the record being + /// acknowledged was almost always just appended; older duplicates (there + /// should not normally be more than one record per key) are left + /// untouched. + pub fn mark_hosted_delivered(&mut self, dedupe_key: &str) -> bool { + for record in self.records.iter_mut().rev() { + if record.hosted_delivery == HostedDeliveryState::Pending + && record.dedupe_key() == dedupe_key + { + record.hosted_delivery = HostedDeliveryState::Delivered; + self.dirty = true; + return true; + } + } + false + } + + /// Whether in-memory state has changed since the last confirmed + /// successful [`save`]/[`persist`](CrashInsights::persist). Exposed + /// mainly for tests; production code should prefer + /// [`take_dirty`](CrashInsights::take_dirty) so a check-and-clear is + /// atomic. + pub fn is_dirty(&self) -> bool { + self.dirty + } + + /// Read and clear the dirty flag. Callers that get `true` back are + /// responsible for attempting a [`persist`](CrashInsights::persist); if + /// that attempt fails, `persist` re-sets the flag itself so the next + /// caller retries. + pub fn take_dirty(&mut self) -> bool { + std::mem::take(&mut self.dirty) + } + + /// Force the dirty flag on, e.g. after an external caller detects a + /// failed write through some other path and wants to guarantee a future + /// retry. + pub fn mark_dirty(&mut self) { + self.dirty = true; + } + + /// Total count of failed [`persist`] attempts since process start. Pure + /// observability, surfaced via the broker's `GetCrashInsights` API so an + /// operator can see durability trouble without having to grep logs. + pub fn save_failures_total(&self) -> u64 { + self.save_failures_total + } + + /// Count of retention-pressure events where the durable pending outbox + /// filled the generic cap and no eviction could happen (see `record`). + /// Operator-visible pressure signal — pairs with + /// `pending_hosted_deliveries().len()` to distinguish "growing but + /// healthy" from "under sustained backpressure." + pub fn retention_pressure_total(&self) -> u64 { + self.retention_pressure_total + } + + /// Attempt to durably persist the current state to `path`, with + /// built-in failure bookkeeping: a successful write clears the dirty + /// flag; a failed write increments [`save_failures_total`] and leaves + /// (or re-sets) the dirty flag so a later call — e.g. from a periodic + /// maintenance flush — retries automatically. This turns a transient + /// write failure into a bounded-retry-until-success operation instead + /// of a silent, one-shot, log-only data loss: the in-memory state (and + /// therefore the next in-process read of it, e.g. via the API) is never + /// wrong, and the on-disk copy is guaranteed another attempt as long as + /// the process keeps calling this on its normal cadence (every + /// maintenance tick). + /// + /// Returns `true` on success, `false` on failure (the error itself is + /// intentionally not returned — callers that want the error text should + /// call [`save`](CrashInsights::save) directly and do their own + /// bookkeeping, as the durable-outbox-critical call sites still do so + /// they can log full context). + pub fn persist(&mut self, path: &Path) -> bool { + match self.save(path) { + Ok(()) => { + self.dirty = false; + true + } + Err(_) => { + self.save_failures_total += 1; + self.dirty = true; + false + } + } + } + /// Load from a JSON file. Returns empty insights if file doesn't exist or is invalid. pub fn load(path: &Path) -> Self { std::fs::read_to_string(path) @@ -186,10 +462,16 @@ impl CrashInsights { /// Save to a JSON file. pub fn save(&self, path: &Path) -> anyhow::Result<()> { let json = serde_json::to_string_pretty(self)?; - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent)?; - } - std::fs::write(path, json)?; + let parent = path.parent().unwrap_or_else(|| Path::new(".")); + std::fs::create_dir_all(parent)?; + // Replace the snapshot atomically. Exit records are written from the + // maintenance tick, so a broker crash during persistence must leave + // either the prior complete snapshot or this complete one, never a + // truncated JSON file that erases all earlier evidence on restart. + let mut tmp = tempfile::NamedTempFile::new_in(parent)?; + std::io::Write::write_all(&mut tmp, json.as_bytes())?; + tmp.as_file().sync_all()?; + tmp.persist(path).map_err(|error| error.error)?; Ok(()) } @@ -200,6 +482,27 @@ impl CrashInsights { "recent": self.recent(20), "patterns": self.patterns(), "health_score": self.health_score(), + // Durable hosted-delivery outbox status. `hosted_delivery_pending` + // counts records not yet handed off to the hosted publisher + // channel — these are exactly what gets replayed on the next + // broker restart. Backward-compatible addition: older clients + // that don't read this field are unaffected. + "hosted_delivery_pending": self.pending_hosted_deliveries().len(), + // Durability status for the on-disk snapshot itself (distinct + // from hosted-delivery status above). Non-zero means at least + // one `persist` write has failed since process start; `dirty == + // true` means the in-memory state is not yet confirmed written + // to disk and a retry is pending on the next maintenance flush. + "save_failures_total": self.save_failures_total, + "dirty": self.dirty, + // Pressure policy visibility (see `record`): diagnostics + // (patterns/health_score/recent above) are intentionally + // bounded/lossy by design — old, already-delivered crash + // history is dropped once `max_records` is exceeded. The + // durable pending outbox is the opposite: it is *never* + // silently evicted under pressure, so this counter increments + // instead whenever eviction was skipped to protect it. + "retention_pressure_total": self.retention_pressure_total, }) } } @@ -218,6 +521,16 @@ mod tests { uptime_secs: 60, category, description, + workspace_id: None, + spawn_invocation_id: None, + generation: String::new(), + became_ready: true, + spawned_at: 0, + ready_at: None, + exited_at: 0, + exit_reason: None, + fleet_node_name: None, + hosted_delivery: HostedDeliveryState::Pending, } } @@ -291,14 +604,24 @@ mod tests { } #[test] - fn records_trimmed_to_max() { + fn records_trimmed_to_max_when_all_delivered() { + // Deliberate retention semantics: the generic cap only ever evicts + // records whose hosted delivery has already succeeded (see + // `retention_bounds_pending_hosted_delivery_backlog` below for the + // pending-preserving half of this contract). With none pending, + // trimming behaves like a plain bounded ring: oldest evicted first. let mut ci = CrashInsights { records: Vec::new(), max_records: 3, + dirty: false, + save_failures_total: 0, + retention_pressure_total: 0, }; for i in 0..5 { - ci.record(make_record(&format!("w{}", i), Some(1), None)); + let mut record = make_record(&format!("w{}", i), Some(1), None); + record.hosted_delivery = HostedDeliveryState::Delivered; + ci.record(record); } assert_eq!(ci.total(), 3); @@ -308,6 +631,36 @@ mod tests { assert_eq!(ci.records[2].agent_name, "w4"); } + #[test] + fn records_over_cap_are_retained_while_pending_hosted_delivery() { + // Deliberate retention semantics (the other half of the contract + // above): the durable pending-delivery outbox must never be + // silently evicted by the generic crash-history cap, even though + // that means the on-disk file can temporarily grow past + // `max_records` while deliveries are outstanding. This is + // documented, bounded-but-not-silently-lossy behavior — see + // `CrashInsights::record` and `pending_hosted_deliveries`. + let mut ci = CrashInsights { + records: Vec::new(), + max_records: 3, + dirty: false, + save_failures_total: 0, + retention_pressure_total: 0, + }; + + for i in 0..5 { + // Default `hosted_delivery` is `Pending` (see `make_record`). + ci.record(make_record(&format!("w{}", i), Some(1), None)); + } + + assert_eq!( + ci.total(), + 5, + "pending hosted-delivery records must survive past the generic retention cap" + ); + assert_eq!(ci.pending_hosted_deliveries().len(), 5); + } + #[test] fn patterns_group_by_category() { let mut ci = CrashInsights::new(); @@ -378,6 +731,54 @@ mod tests { assert_eq!(loaded.records[0].agent_name, "w1"); } + #[test] + fn correlated_exit_metadata_survives_atomic_persistence() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("crashes.json"); + let mut ci = CrashInsights::new(); + let mut record = make_record("w1", Some(137), None); + record.workspace_id = Some("ws-1".to_string()); + record.spawn_invocation_id = Some("invoke-1".to_string()); + record.generation = "generation-1".to_string(); + record.became_ready = true; + record.spawned_at = 100; + record.ready_at = Some(110); + record.exited_at = 125; + record.uptime_secs = 25; + record.exit_reason = Some("worker_write_failed".to_string()); + record.fleet_node_name = Some("node-1".to_string()); + ci.record(record); + + ci.save(&path).unwrap(); + let loaded = CrashInsights::load(&path); + let loaded = &loaded.records[0]; + assert_eq!(loaded.spawn_invocation_id.as_deref(), Some("invoke-1")); + assert_eq!(loaded.generation, "generation-1"); + assert_eq!(loaded.ready_at, Some(110)); + assert_eq!(loaded.exited_at, 125); + assert_eq!(loaded.exit_reason.as_deref(), Some("worker_write_failed")); + } + + #[test] + fn legacy_records_load_with_empty_correlation_fields() { + let record: CrashRecord = serde_json::from_value(serde_json::json!({ + "agent_name": "legacy", + "exit_code": 1, + "signal": null, + "timestamp": 42, + "uptime_secs": 2, + "category": "error", + "description": "Exited with code 1" + })) + .unwrap(); + + assert_eq!(record.agent_name, "legacy"); + assert!(record.generation.is_empty()); + assert!(!record.became_ready); + assert_eq!(record.exited_at, 0); + assert!(record.exit_reason.is_none()); + } + #[test] fn load_missing_file_returns_empty() { let ci = CrashInsights::load(Path::new("/nonexistent/crashes.json")); @@ -412,6 +813,314 @@ mod tests { } } + #[test] + fn new_record_defaults_to_pending_hosted_delivery() { + let mut ci = CrashInsights::new(); + let mut record = make_record("w1", Some(1), None); + record.generation = "gen-1".to_string(); + ci.record(record); + + assert_eq!(ci.pending_hosted_deliveries().len(), 1); + assert_eq!( + ci.pending_hosted_deliveries()[0].hosted_delivery, + HostedDeliveryState::Pending + ); + } + + #[test] + fn legacy_records_without_the_field_default_to_pending() { + // A record persisted by a broker built before `hosted_delivery` + // existed must still be replayed, never silently treated as + // already delivered. + let record: CrashRecord = serde_json::from_value(serde_json::json!({ + "agent_name": "legacy", + "exit_code": 1, + "signal": null, + "timestamp": 42, + "uptime_secs": 2, + "category": "error", + "description": "Exited with code 1", + "generation": "gen-legacy" + })) + .unwrap(); + + assert_eq!(record.hosted_delivery, HostedDeliveryState::Pending); + } + + #[test] + fn mark_hosted_delivered_flips_state_and_returns_true() { + let mut ci = CrashInsights::new(); + let mut record = make_record("w1", Some(1), None); + record.generation = "gen-1".to_string(); + let key = record.dedupe_key(); + ci.record(record); + + assert!(ci.mark_hosted_delivered(&key)); + assert_eq!(ci.pending_hosted_deliveries().len(), 0); + // Marking again finds no pending match — already delivered. + assert!(!ci.mark_hosted_delivered(&key)); + } + + #[test] + fn mark_hosted_delivered_ignores_unknown_key() { + let mut ci = CrashInsights::new(); + let mut record = make_record("w1", Some(1), None); + record.generation = "gen-1".to_string(); + ci.record(record); + + assert!(!ci.mark_hosted_delivered("w1::some-other-generation")); + assert_eq!(ci.pending_hosted_deliveries().len(), 1); + } + + #[test] + fn dedupe_key_distinguishes_same_name_different_generation() { + let mut a = make_record("w1", Some(1), None); + a.generation = "gen-old".to_string(); + let mut b = make_record("w1", Some(1), None); + b.generation = "gen-new".to_string(); + + assert_ne!(a.dedupe_key(), b.dedupe_key()); + + // Marking the old generation delivered must not affect the new + // generation's own pending record — a same-name replacement worker's + // exit must remain independently deliverable. + let mut ci = CrashInsights::new(); + let old_key = a.dedupe_key(); + ci.record(a); + ci.record(b); + assert!(ci.mark_hosted_delivered(&old_key)); + assert_eq!(ci.pending_hosted_deliveries().len(), 1); + assert_eq!(ci.pending_hosted_deliveries()[0].generation, "gen-new"); + } + + #[test] + fn pending_hosted_deliveries_preserve_chronological_order() { + let mut ci = CrashInsights::new(); + for i in 0..5 { + let mut record = make_record(&format!("w{}", i), Some(1), None); + record.generation = format!("gen-{}", i); + ci.record(record); + } + + let pending = ci.pending_hosted_deliveries(); + let names: Vec<&str> = pending.iter().map(|r| r.agent_name.as_str()).collect(); + assert_eq!(names, vec!["w0", "w1", "w2", "w3", "w4"]); + } + + #[test] + fn retention_bounds_pending_hosted_delivery_backlog() { + // Explicit retention bound: once the generic crash history is full, + // delivered records are evicted before pending hosted exits so the + // durable outbox survives a prolonged outage. + let mut ci = CrashInsights { + records: Vec::new(), + max_records: 3, + dirty: false, + save_failures_total: 0, + retention_pressure_total: 0, + }; + let mut delivered = make_record("delivered", Some(1), None); + delivered.generation = "gen-delivered".to_string(); + delivered.hosted_delivery = HostedDeliveryState::Delivered; + ci.record(delivered); + for i in 0..5 { + let mut record = make_record(&format!("w{}", i), Some(1), None); + record.generation = format!("gen-{}", i); + ci.record(record); + } + + assert_eq!(ci.total(), 5); + assert_eq!(ci.pending_hosted_deliveries().len(), 5); + let names: Vec<&str> = ci + .pending_hosted_deliveries() + .iter() + .map(|r| r.agent_name.as_str()) + .collect(); + assert_eq!(names, vec!["w0", "w1", "w2", "w3", "w4"]); + assert!(ci + .recent(5) + .iter() + .all(|record| record.hosted_delivery == HostedDeliveryState::Pending)); + } + + #[test] + fn hosted_delivery_state_survives_atomic_persistence_round_trip() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("crashes.json"); + let mut ci = CrashInsights::new(); + let mut delivered = make_record("delivered-agent", Some(1), None); + delivered.generation = "gen-delivered".to_string(); + let delivered_key = delivered.dedupe_key(); + ci.record(delivered); + ci.mark_hosted_delivered(&delivered_key); + + let mut pending = make_record("pending-agent", Some(1), None); + pending.generation = "gen-pending".to_string(); + ci.record(pending); + + ci.save(&path).unwrap(); + let reloaded = CrashInsights::load(&path); + + assert_eq!(reloaded.pending_hosted_deliveries().len(), 1); + assert_eq!( + reloaded.pending_hosted_deliveries()[0].agent_name, + "pending-agent" + ); + } + + #[test] + fn dedupe_key_and_persistence_handle_utf8_agent_names() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("crashes.json"); + let mut ci = CrashInsights::new(); + let mut record = make_record("agent-日本語-🚀", Some(1), None); + record.generation = "gen-utf8".to_string(); + let key = record.dedupe_key(); + assert_eq!(key, "agent-日本語-🚀::gen-utf8"); + ci.record(record); + + ci.save(&path).unwrap(); + let mut reloaded = CrashInsights::load(&path); + assert_eq!(reloaded.pending_hosted_deliveries().len(), 1); + assert!(reloaded.mark_hosted_delivered(&key)); + reloaded.save(&path).unwrap(); + + let reloaded_again = CrashInsights::load(&path); + assert_eq!(reloaded_again.pending_hosted_deliveries().len(), 0); + } + + #[test] + fn retention_pressure_counter_increments_when_pending_outbox_blocks_eviction() { + // Documented pressure policy: when every record above the generic + // cap is a still-pending durable outbox entry, `record` must not + // silently evict any of them — instead it must count the pressure + // event so operators can see it (see `retention_pressure_total` + // doc comment and the module-level storage-tradeoff docs above). + let mut ci = CrashInsights { + records: Vec::new(), + max_records: 3, + dirty: false, + save_failures_total: 0, + retention_pressure_total: 0, + }; + + for i in 0..6 { + let mut record = make_record(&format!("w{}", i), Some(1), None); + record.generation = format!("gen-{}", i); + ci.record(record); + } + + assert_eq!( + ci.total(), + 6, + "pending outbox records must never be silently evicted under pressure" + ); + assert_eq!(ci.pending_hosted_deliveries().len(), 6); + // Pressure fires once per `record` call once the cap is exceeded + // and nothing evictable is found: records 4, 5, and 6 (indices 3..6) + // each hit the cap with zero non-pending candidates. + assert_eq!( + ci.retention_pressure_total(), + 3, + "every record call that could not evict anything must count as pressure" + ); + + // Once a delivery is acknowledged, the evictable record is removed + // (never counted as pressure for that removal); pressure only fires + // again for the remaining, still-all-pending excess above the cap. + let delivered_key = ci.records[0].dedupe_key(); + assert!(ci.mark_hosted_delivered(&delivered_key)); + let mut record = make_record("w6", Some(1), None); + record.generation = "gen-6".to_string(); + ci.record(record); + assert_eq!( + ci.retention_pressure_total(), + 4, + "the one evictable (delivered) record is removed for free; pressure fires \ + again only for the remaining pending excess above the cap" + ); + assert_eq!(ci.pending_hosted_deliveries().len(), 6); + assert_eq!( + ci.total(), + 6, + "the delivered record was evicted, keeping total at 6" + ); + } + + #[test] + fn persist_failure_increments_counter_and_stays_dirty() { + // Deterministic forced-failure fixture: `parent` is a *file*, not a + // directory, so `std::fs::create_dir_all(parent)` inside `save` + // fails every time on every platform — no flaky IO mocking needed. + let dir = tempfile::tempdir().unwrap(); + let not_a_dir = dir.path().join("not-a-directory"); + std::fs::write(¬_a_dir, b"blocking file").unwrap(); + let path = not_a_dir.join("crashes.json"); + + let mut ci = CrashInsights::new(); + ci.record(make_record("w1", Some(1), None)); + assert!(ci.is_dirty()); + assert_eq!(ci.save_failures_total(), 0); + + assert!(!ci.persist(&path), "forced-failure path must fail"); + assert_eq!( + ci.save_failures_total(), + 1, + "a failed persist must be counted, not merely logged" + ); + assert!( + ci.is_dirty(), + "a failed persist must leave (or re-set) the dirty flag so a later flush retries" + ); + + // A second failed attempt keeps incrementing and stays dirty — + // durability is never silently given up on. + assert!(!ci.persist(&path)); + assert_eq!(ci.save_failures_total(), 2); + assert!(ci.is_dirty()); + } + + #[test] + fn persist_recovery_clears_dirty_and_stops_incrementing() { + let dir = tempfile::tempdir().unwrap(); + let not_a_dir = dir.path().join("not-a-directory"); + std::fs::write(¬_a_dir, b"blocking file").unwrap(); + let bad_path = not_a_dir.join("crashes.json"); + let good_path = dir.path().join("crashes.json"); + + let mut ci = CrashInsights::new(); + ci.record(make_record("w1", Some(1), None)); + + // Fail twice against the unwritable path. + assert!(!ci.persist(&bad_path)); + assert!(!ci.persist(&bad_path)); + assert_eq!(ci.save_failures_total(), 2); + assert!(ci.is_dirty()); + + // Recovery: the next attempt against a writable path succeeds, + // clears the dirty flag, and does not touch the failure counter. + assert!(ci.persist(&good_path)); + assert_eq!( + ci.save_failures_total(), + 2, + "a successful persist must not increment the failure counter" + ); + assert!( + !ci.is_dirty(), + "a confirmed-successful persist must clear the dirty flag" + ); + + // take_dirty() reflects the cleared state and further successful + // persists remain no-ops on the counter. + assert!(!ci.take_dirty()); + assert!(ci.persist(&good_path)); + assert_eq!(ci.save_failures_total(), 2); + + let loaded = CrashInsights::load(&good_path); + assert_eq!(loaded.total(), 1); + assert_eq!(loaded.records[0].agent_name, "w1"); + } + #[test] fn recent_returns_most_recent() { let mut ci = CrashInsights::new(); diff --git a/packages/harness-driver/src/protocol.ts b/packages/harness-driver/src/protocol.ts index 0e82a7ea85..c704f80997 100644 --- a/packages/harness-driver/src/protocol.ts +++ b/packages/harness-driver/src/protocol.ts @@ -355,6 +355,15 @@ export interface CrashRecord { uptime_secs: number; category: CrashCategory; description: string; + workspace_id?: string; + spawn_invocation_id?: string; + generation?: string; + became_ready?: boolean; + spawned_at?: number; + ready_at?: number; + exited_at?: number; + exit_reason?: string; + fleet_node_name?: string; } export interface CrashPattern { @@ -407,6 +416,13 @@ export type BrokerEvent = signal?: string; reason?: string; generation?: string; + workspace_id?: string; + spawn_invocation_id?: string; + fleet_node_name?: string; + became_ready?: boolean; + spawned_at?: number; + ready_at?: number; + exited_at?: number; } | { kind: 'agent_context_low'; diff --git a/tests/e2e/fleet/get-agent-retry.test.ts b/tests/e2e/fleet/get-agent-retry.test.ts new file mode 100644 index 0000000000..a189c7f973 --- /dev/null +++ b/tests/e2e/fleet/get-agent-retry.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from 'vitest'; +import { + GET_AGENT_RATE_LIMIT_DEFAULT_BACKOFF_MS, + GET_AGENT_RATE_LIMIT_MAX_ATTEMPTS, + getAgent, + type EngineHandle, +} from './harness.js'; + +/** Minimal fake `EngineHandle` whose `fetchJson` is driven by a scripted list + * of responses (one per call), so `getAgent`'s retry logic can be exercised + * deterministically without a real Relaycast engine or network. */ +function fakeEngine(responses: Array<{ status: number; body: unknown; headers?: Record }>): { + engine: EngineHandle; + calls: number[]; +} { + const calls: number[] = []; + let index = 0; + const engine: EngineHandle = { + baseUrl: 'http://127.0.0.1:0', + port: 0, + async stop() {}, + async fetchJson() { + calls.push(Date.now()); + const next = responses[Math.min(index, responses.length - 1)]; + index += 1; + return { + status: next.status, + body: next.body, + headers: new Headers(next.headers ?? {}), + }; + }, + }; + return { engine, calls }; +} + +const RATE_LIMIT_BODY = { + ok: false, + error: { + code: 'rate_limit_exceeded', + message: 'Rate limit exceeded. 300 requests per minute allowed for free plan.', + }, +}; + +describe('getAgent 429 rate_limit_exceeded retry', () => { + it('retries a single rate_limit_exceeded 429 honoring Retry-After and eventually succeeds', async () => { + const { engine, calls } = fakeEngine([ + { status: 429, body: RATE_LIMIT_BODY, headers: { 'retry-after': '0' } }, + { status: 200, body: { data: { name: 'worker-a' } } }, + ]); + + const started = Date.now(); + const result = await getAgent(engine, 'rk_test', 'worker-a'); + const elapsedMs = Date.now() - started; + + expect(result).toEqual({ name: 'worker-a' }); + expect(calls).toHaveLength(2); + // Retry-After: 0 means no meaningful delay is required. + expect(elapsedMs).toBeLessThan(1_000); + }); + + it('exhausts the attempt/deadline cap and throws on a persistent rate_limit_exceeded 429', async () => { + const { engine, calls } = fakeEngine([ + { status: 429, body: RATE_LIMIT_BODY, headers: { 'retry-after': '0' } }, + ]); + + await expect(getAgent(engine, 'rk_test', 'worker-a')).rejects.toThrow(/rate_limit_exceeded/); + expect(calls.length).toBe(GET_AGENT_RATE_LIMIT_MAX_ATTEMPTS); + }); + + it('fails fast (does not hang) when the rate limit persists without Retry-After', async () => { + const { engine } = fakeEngine([{ status: 429, body: RATE_LIMIT_BODY }]); + + const started = Date.now(); + await expect(getAgent(engine, 'rk_test', 'worker-a')).rejects.toThrow(); + const elapsedMs = Date.now() - started; + + // Default backoff is bounded and small; the whole retry budget must stay + // well under the deadline cap, not hang. + const worstCase = GET_AGENT_RATE_LIMIT_DEFAULT_BACKOFF_MS * GET_AGENT_RATE_LIMIT_MAX_ATTEMPTS + 1_000; + expect(elapsedMs).toBeLessThan(worstCase); + }); + + it('does not retry a non-rate_limit_exceeded 429 and throws immediately', async () => { + const { engine, calls } = fakeEngine([ + { status: 429, body: { ok: false, error: { code: 'some_other_error', message: 'nope' } } }, + ]); + + await expect(getAgent(engine, 'rk_test', 'worker-a')).rejects.toThrow(/429/); + expect(calls).toHaveLength(1); + }); + + it('does not retry other statuses and throws immediately, exactly as before', async () => { + const { engine, calls } = fakeEngine([{ status: 500, body: { ok: false } }]); + + await expect(getAgent(engine, 'rk_test', 'worker-a')).rejects.toThrow(/500/); + expect(calls).toHaveLength(1); + }); + + it('still returns null on 404 without retrying', async () => { + const { engine, calls } = fakeEngine([{ status: 404, body: {} }]); + + const result = await getAgent(engine, 'rk_test', 'worker-a'); + expect(result).toBeNull(); + expect(calls).toHaveLength(1); + }); +}); diff --git a/tests/e2e/fleet/harness.ts b/tests/e2e/fleet/harness.ts index 673e10e234..575571de16 100644 --- a/tests/e2e/fleet/harness.ts +++ b/tests/e2e/fleet/harness.ts @@ -160,7 +160,7 @@ export interface EngineHandle { baseUrl: string; port: number; stop(): Promise; - fetchJson(pathname: string, init?: RequestInit): Promise<{ status: number; body: any }>; + fetchJson(pathname: string, init?: RequestInit): Promise<{ status: number; body: any; headers: Headers }>; } export async function startEngine( @@ -183,7 +183,7 @@ export async function startEngine( const fetchJson = async (pathname: string, init: RequestInit = {}) => { const res = await fetch(`${baseUrl}${pathname}`, init); - return { status: res.status, body: await res.json().catch(() => ({})) }; + return { status: res.status, body: await res.json().catch(() => ({})), headers: res.headers }; }; await waitFor( @@ -670,24 +670,95 @@ export async function listMessages( return items as Array<{ text: string }>; } +/** Bound on the number of `getAgent` attempts (including the first) that + * `rate_limit_exceeded` 429s may be retried before giving up. Exported for + * tests. */ +export const GET_AGENT_RATE_LIMIT_MAX_ATTEMPTS = 4; + +/** Bound on the total wall-clock time `getAgent` may spend retrying + * `rate_limit_exceeded` 429s before giving up, regardless of how many + * attempts remain. Exported for tests. */ +export const GET_AGENT_RATE_LIMIT_MAX_WAIT_MS = 5_000; + +/** Fallback backoff (ms) used when a `rate_limit_exceeded` 429 response + * carries no usable `Retry-After`. Exported for tests. */ +export const GET_AGENT_RATE_LIMIT_DEFAULT_BACKOFF_MS = 250; + +function isRateLimitExceededBody(body: unknown): boolean { + return ( + typeof body === 'object' && + body !== null && + (body as { ok?: unknown }).ok === false && + typeof (body as { error?: { code?: unknown } }).error === 'object' && + (body as { error?: { code?: unknown } }).error !== null && + (body as { error: { code?: unknown } }).error.code === 'rate_limit_exceeded' + ); +} + +/** Parse a `Retry-After` value (header or mirrored body field) into a + * millisecond backoff. Supports the HTTP-standard delta-seconds form; any + * other value (including an HTTP-date, which this helper does not attempt + * to parse) is treated as absent so callers fall back to a deterministic + * default rather than guessing. Returns `null` when no usable value is + * present. */ +function parseRetryAfterMs(retryAfter: string | null | undefined): number | null { + if (!retryAfter) return null; + const seconds = Number(retryAfter); + if (!Number.isFinite(seconds) || seconds < 0) return null; + return seconds * 1000; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + /** Read one agent's engine record, including the metadata bag. * * Returns null ONLY for 404 — "the agent does not exist yet", which callers * poll on. Every other failure throws: mapping auth, server, and not-found * errors all to null makes a broken read indistinguishable from a legitimately * absent agent, and a `waitFor` polling on null would then time out (or an - * assertion would pass) for entirely the wrong reason. */ + * assertion would pass) for entirely the wrong reason. + * + * The one exception is a `429 rate_limit_exceeded` response (the free-plan + * Relaycast quota): that specific, transient shape is retried with a + * bounded, deterministic backoff (honoring a valid `Retry-After` when + * present, no jitter) capped at both `GET_AGENT_RATE_LIMIT_MAX_ATTEMPTS` + * attempts and `GET_AGENT_RATE_LIMIT_MAX_WAIT_MS` total wall-clock time, so a + * persistent rate limit still fails fast instead of hanging. Any other 429 + * error code, or any other status, throws immediately exactly as before. */ export async function getAgent( engine: EngineHandle, workspaceKey: string, name: string ): Promise<{ name: string; metadata?: Record } | null> { - const { status, body } = await engine.fetchJson(`/v1/agents/${name}`, { - headers: { authorization: `Bearer ${workspaceKey}` }, - }); - if (status === 404) return null; - if (status >= 300) throw new Error(`getAgent(${name}) ${status}: ${JSON.stringify(body)}`); - return body.data ?? null; + const deadline = Date.now() + GET_AGENT_RATE_LIMIT_MAX_WAIT_MS; + let attempt = 0; + for (;;) { + attempt += 1; + const { status, body, headers } = await engine.fetchJson(`/v1/agents/${name}`, { + headers: { authorization: `Bearer ${workspaceKey}` }, + }); + if (status === 404) return null; + if (status === 429 && isRateLimitExceededBody(body)) { + const remainingMs = deadline - Date.now(); + if (attempt >= GET_AGENT_RATE_LIMIT_MAX_ATTEMPTS || remainingMs <= 0) { + throw new Error( + `getAgent(${name}) ${status}: ${JSON.stringify(body)} (gave up after ${attempt} attempts)` + ); + } + const retryAfterMs = + parseRetryAfterMs(headers?.get?.('retry-after')) ?? + parseRetryAfterMs( + (body as { error?: { retry_after?: unknown } }).error?.retry_after as string | undefined + ) ?? + GET_AGENT_RATE_LIMIT_DEFAULT_BACKOFF_MS; + await sleep(Math.max(0, Math.min(retryAfterMs, remainingMs))); + continue; + } + if (status >= 300) throw new Error(`getAgent(${name}) ${status}: ${JSON.stringify(body)}`); + return body.data ?? null; + } } /** Release (delete) an agent, freeing its location — used to model a resumable diff --git a/tests/relayflows/cases/1603-hosted-exit-durable-delivery/case.json b/tests/relayflows/cases/1603-hosted-exit-durable-delivery/case.json new file mode 100644 index 0000000000..18b3ff6b5b --- /dev/null +++ b/tests/relayflows/cases/1603-hosted-exit-durable-delivery/case.json @@ -0,0 +1,21 @@ +{ + "version": 1, + "id": "1603-hosted-exit-durable-delivery", + "kind": "bugfix", + "title": "Persist and restart-replay a real fleet-spawned child exit", + "requirements": ["broker-linux-x64"], + "runner": { + "command": ["node", "tests/relayflows/cases/1603-hosted-exit-durable-delivery/run.mjs"] + }, + "timeoutSeconds": 900, + "expected": { + "base": { + "outcome": "bug", + "signature": "live_exit_never_enters_durable_hosted_outbox" + }, + "head": { + "outcome": "fixed", + "signature": "live_fleet_child_exit_persists_then_replays_once_after_restart" + } + } +} diff --git a/tests/relayflows/cases/1603-hosted-exit-durable-delivery/run.mjs b/tests/relayflows/cases/1603-hosted-exit-durable-delivery/run.mjs new file mode 100644 index 0000000000..4c434f03b7 --- /dev/null +++ b/tests/relayflows/cases/1603-hosted-exit-durable-delivery/run.mjs @@ -0,0 +1,565 @@ +// Proves #1603 through the real public fleet-control path. The local fake is +// deliberately transport-only: HTTP registration/events plus /v1/node/ws. +// It drives action.invoke, launches a real disposable child, holds the first +// event request open after persistence, kills the broker, then enables HTTP +// success for the restarted broker's replay. +// +// >256 backlog replenishment and retention pressure remain unit-level proof in +// runtime/tests.rs and crash_insights.rs; this live case does not claim them. +import { createHash, randomUUID } from 'node:crypto'; +import { execFileSync, spawn } from 'node:child_process'; +import http from 'node:http'; +import { constants as fsConstants } from 'node:fs'; +import { access, mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import process from 'node:process'; +import { fileURLToPath } from 'node:url'; + +const CASE_ID = '1603-hosted-exit-durable-delivery'; +const AGENT_NAME = 'relayflow-1603-live-child'; +const INSTANCE_NAME = 'relayflow-1603-live-broker'; +const INVOCATION_ID = 'relayflow-1603-live-invocation'; +const EXIT_WINDOW_MS = 15_000; +const REPLAY_WINDOW_MS = 15_000; +const targetDir = requiredDirectory('RELAY_PR_PROOF_TARGET_DIR'); +const harnessDir = requiredDirectory('RELAY_PR_PROOF_HARNESS_DIR'); +const binaryPath = await requiredExecutable('RELAY_PR_PROOF_BROKER_BINARY'); +const resultPath = requiredValue('RELAY_PR_PROOF_RESULT_PATH'); +const arm = requiredValue('RELAY_PR_PROOF_ARM'); + +if (!['base', 'head'].includes(arm)) throw new Error(`Unexpected proof arm ${JSON.stringify(arm)}.`); +const expectedSha = + arm === 'base' ? process.env.RELAY_PR_PROOF_BASE_SHA : process.env.RELAY_PR_PROOF_HEAD_SHA; +if (!expectedSha) throw new Error(`Missing expected ${arm} SHA.`); +const targetSha = execFileSync('git', ['-C', targetDir, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim(); +if (targetSha !== expectedSha) + throw new Error(`Target checkout ${targetSha} does not match exact ${arm} SHA ${expectedSha}.`); +if (!isWithin(harnessDir, fileURLToPath(import.meta.url))) + throw new Error('Runner must execute from the exact-head harness checkout.'); + +const probeDir = await mkdtemp(path.join(tmpdir(), 'relayflow-1603-live-')); +const stateDir = path.join(probeDir, 'state'); +const childExitProofPath = path.join(probeDir, 'real-child-exit-code'); +const childProofNonce = randomUUID(); +const childSessionRef = `relayflow-1603-live-session-${childProofNonce}`; +let relaycast; +let firstBroker; +let replayBroker; +let successfulSpawnActionResult; +let realChildExit; +let realChildExitProof; +let brokerChildLifecycle; +try { + relaycast = await startFakeRelaycast({ + childExitProofPath, + childProofNonce, + childSessionRef, + }); + const env = { + PATH: process.env.PATH ?? '/usr/bin:/bin', + HOME: probeDir, + TMPDIR: probeDir, + NO_COLOR: '1', + RELAYCAST_BASE_URL: relaycast.baseUrl, + RELAY_NODE_TOKEN: 'nt_relayflow_1603', + RELAY_NODE_ID: 'node_relayflow_1603', + AGENT_RELAY_WORKSPACE_KEY: 'rk_relayflow_1603', + AGENT_RELAY_STARTUP_DEBUG: '1', + AGENT_RELAY_TELEMETRY_DISABLED: '1', + }; + firstBroker = startBroker({ binaryPath, cwd: probeDir, stateDir, env }); + await waitFor( + () => relaycast.observations().spawnRequests === 1, + EXIT_WINDOW_MS, + 'fleet action.invoke was not sent' + ); + await waitFor( + () => relaycast.successfulSpawnActionResult(), + EXIT_WINDOW_MS, + 'fleet spawn never returned a successful action.result for the invocation' + ); + successfulSpawnActionResult = relaycast.successfulSpawnActionResult(); + await waitFor( + async () => { + realChildExit = await expectedChildExit(stateDir); + realChildExitProof = await childExitProof(childExitProofPath, childProofNonce, childSessionRef); + brokerChildLifecycle = relaycast.expectedChildLifecycle(); + return Boolean(realChildExit || realChildExitProof) && brokerChildLifecycle; + }, + EXIT_WINDOW_MS, + 'the fleet-spawned child never recorded the expected exit code 23' + ); + await waitFor( + () => relaycast.observations().eventAttempts === 1, + EXIT_WINDOW_MS, + 'real child exit never reached Relaycast publication' + ); + const beforeRestart = await crashInsights(stateDir); + const pending = beforeRestart.records.find((record) => record.agent_name === AGENT_NAME); + if (!expectedChildExitRecord(pending) || pending.hosted_delivery !== 'pending' || !pending.generation) { + throw diagnostic('The real pre-restart exit was not durably Pending.', { + beforeRestart, + pending, + relaycast: relaycast.observations(), + firstBroker, + }); + } + + // The first request is deliberately unanswered: persistence has happened, + // but no HTTP success exists. Cross a real process restart boundary now. + await stopBroker(firstBroker, 'SIGKILL'); + firstBroker = undefined; + await relaycast.enableDelivery(); + replayBroker = startBroker({ binaryPath, cwd: probeDir, stateDir, env }); + await waitFor( + () => relaycast.observations().eventAttempts === 2, + REPLAY_WINDOW_MS, + 'restart did not replay the pending hosted exit' + ); + await waitFor( + async () => { + const state = await crashInsights(stateDir); + return state.records.some( + (record) => + expectedChildExitRecord(record) && + record.generation === pending.generation && + record.hosted_delivery === 'delivered' + ); + }, + REPLAY_WINDOW_MS, + 'HTTP 200 did not mark the real exit Delivered' + ); + const final = relaycast.observations(); + const afterRestart = await crashInsights(stateDir); + const expectedDedupeKey = `${AGENT_NAME}::${pending.generation}`; + const dedupeKeys = final.eventBodies.map((body) => body?.payload?.dedupe_key); + if ( + final.spawnRequests !== 1 || + final.eventAttempts !== 2 || + !successfulSpawnActionResult || + !brokerChildLifecycle || + !final.eventBodies.every((body) => + expectedAgentExitedEvent(body, pending.generation, expectedDedupeKey) + ) || + !dedupeKeys.every((key) => key === expectedDedupeKey) + ) { + throw diagnostic('Unexpected live fleet-control replay observation.', { + final, + beforeRestart, + afterRestart, + firstBroker, + replayBroker, + }); + } + await writeResult( + 'fixed', + 'live_fleet_child_exit_persists_then_replays_once_after_restart', + `A public /v1/node/ws action.invoke returned successful action.result, launched a real disposable child, and recorded its exit code 23 as Pending before Relaycast HTTP was available. After killing and restarting the exact broker on the same state, it replayed exactly the persisted generation as agent_exited with dedupe_key ${expectedDedupeKey} and became Delivered only after a fake Relaycast HTTP 200.` + ); +} catch (error) { + // Pre-fix binaries legitimately lack the durable-outbox code, so they do + // not make the first event request. They may also lack the durable crash + // record; in that arm only, accept the nonce-bound, child-owned exit marker + // after the exact action result and broker lifecycle prove the same invocation. + if ( + arm === 'base' && + relaycast?.observations().eventAttempts === 0 && + relaycast.observations().spawnRequests === 1 && + successfulSpawnActionResult && + brokerChildLifecycle && + (expectedChildExitRecord(realChildExit) || realChildExitProof) + ) { + await writeResult( + 'bug', + 'live_exit_never_enters_durable_hosted_outbox', + 'The base broker returned successful action.result for the public fleet spawn and the real disposable child proved exit code 23, but the broker never published its hosted exit.' + ); + } else { + throw diagnostic(error.message, { + relaycast: relaycast?.observations(), + firstBroker: firstBroker?.logs(), + replayBroker: replayBroker?.logs(), + }); + } +} finally { + await stopBroker(firstBroker, 'SIGKILL'); + await stopBroker(replayBroker, 'SIGKILL'); + await relaycast?.close(); + await rm(probeDir, { recursive: true, force: true }); +} + +function startBroker({ binaryPath, cwd, stateDir, env }) { + const child = spawn( + binaryPath, + ['init', '--instance-name', INSTANCE_NAME, '--channels', 'general', '--persist', '--state-dir', stateDir], + { cwd, env, stdio: ['ignore', 'pipe', 'pipe'] } + ); + let stdout = ''; + let stderr = ''; + child.stdout.on('data', (chunk) => { + stdout += chunk; + }); + child.stderr.on('data', (chunk) => { + stderr += chunk; + }); + return { child, logs: () => ({ stdout: stdout.slice(-4_000), stderr: stderr.slice(-4_000) }) }; +} +async function stopBroker(broker, signal) { + if (!broker?.child || broker.child.exitCode !== null) return; + broker.child.kill(signal); + await Promise.race([new Promise((resolve) => broker.child.once('exit', resolve)), sleep(5_000)]); + if (broker.child.exitCode === null) broker.child.kill('SIGKILL'); +} +async function crashInsights(directory) { + return JSON.parse(await readFile(path.join(directory, 'crash-insights.json'), 'utf8')); +} +async function expectedChildExit(directory) { + try { + const insights = await crashInsights(directory); + return insights.records.find(expectedChildExitRecord); + } catch (error) { + if (error?.code === 'ENOENT') return undefined; + throw error; + } +} +async function childExitProof(proofPath, nonce, sessionRef) { + try { + const marker = JSON.parse(await readFile(proofPath, 'utf8')); + return ( + marker?.nonce === nonce && + marker.invocation_id === INVOCATION_ID && + marker.session_ref === sessionRef && + marker.status === 23 && + Number.isSafeInteger(marker.pid) && + marker.pid > 1 + ); + } catch (error) { + if (error?.code === 'ENOENT') return false; + throw error; + } +} +function expectedChildExitRecord(record) { + return ( + record?.agent_name === AGENT_NAME && + record.spawn_invocation_id === INVOCATION_ID && + record.exit_code === 23 + ); +} +function expectedAgentExitedEvent(body, generation, dedupeKey) { + return ( + body?.type === 'agent_exited' && + hasExactKeys(body, ['type', 'payload']) && + hasExactKeys(body.payload, [ + 'code', + 'signal', + 'reason', + 'generation', + 'workspace_id', + 'spawn_invocation_id', + 'fleet_node_name', + 'became_ready', + 'spawned_at', + 'ready_at', + 'exited_at', + 'dedupe_key', + ]) && + body.payload?.spawn_invocation_id === INVOCATION_ID && + body.payload?.code === 23 && + body.payload?.generation === generation && + body.payload?.dedupe_key === dedupeKey + ); +} +function hasExactKeys(value, expectedKeys) { + return ( + value && + typeof value === 'object' && + !Array.isArray(value) && + Object.keys(value).length === expectedKeys.length && + expectedKeys.every((key) => Object.hasOwn(value, key)) + ); +} +async function writeResult(outcome, signature, details) { + await mkdir(path.dirname(resultPath), { recursive: true }); + await writeFile( + resultPath, + `${JSON.stringify({ version: 1, caseId: CASE_ID, arm, outcome, signature, details })}\n`, + 'utf8' + ); +} + +/** Minimal Relaycast HTTP plus raw WebSocket fake; no production test hooks. */ +async function startFakeRelaycast({ childExitProofPath, childProofNonce, childSessionRef }) { + let spawnRequests = 0; + let actionSent = false; + let deliveryAvailable = false; + const heldResponses = new Set(); + const eventBodies = []; + const controlMessages = []; + const actionResults = []; + const sockets = new Set(); + const server = http.createServer(async (request, response) => { + const body = await requestBody(request); + if (request.method === 'POST' && request.url === '/v1/agents') { + return json(response, 200, { + ok: true, + data: { + id: 'a_relayflow_broker', + workspace_id: 'ws_relayflow_1603', + name: INSTANCE_NAME, + token: 'at_relayflow_broker', + status: 'online', + created_at: '2025-01-01T00:00:00Z', + }, + }); + } + if (request.method === 'GET' && request.url === `/v1/agents/${AGENT_NAME}`) { + return json(response, 200, { + ok: true, + data: { id: 'a_relayflow_live_child', name: AGENT_NAME, channels: [] }, + }); + } + if (request.method === 'POST' && request.url === `/v1/agents/${AGENT_NAME}/events`) { + let parsed; + try { + parsed = JSON.parse(body); + } catch { + parsed = null; + } + eventBodies.push(parsed); + if (!deliveryAvailable) { + heldResponses.add(response); + response.once('close', () => heldResponses.delete(response)); + return; + } + return json(response, 200, { + ok: true, + data: { + id: 'evt_relayflow_1603', + agent_id: 'a_relayflow_live_child', + type: 'agent_exited', + payload: parsed?.payload ?? {}, + created_at: '2025-01-01T00:00:01Z', + }, + }); + } + return json(response, 404, { ok: false, error: { code: 'not_found', message: request.url } }); + }); + server.on('upgrade', (request, socket) => { + if (!request.url?.startsWith('/v1/node/ws')) return socket.destroy(); + const key = request.headers['sec-websocket-key']; + if (typeof key !== 'string') return socket.destroy(); + const accept = createHash('sha1').update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64'); + socket.write( + `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n` + ); + sockets.add(socket); + let buffer = Buffer.alloc(0); + socket.on('data', (chunk) => { + buffer = Buffer.concat([buffer, chunk]); + let frame; + while ((frame = takeFrame(buffer))) { + buffer = buffer.subarray(frame.consumed); + if (frame.opcode === 8) return socket.end(); + if (frame.opcode === 9) { + socket.write(serverFrame(10, frame.payload)); + continue; + } + if (frame.opcode !== 1) continue; + let message; + try { + message = JSON.parse(frame.payload.toString('utf8')); + } catch { + continue; + } + controlMessages.push(message); + if (message.type === 'action.result') actionResults.push(message); + if (message.type === 'agent.register') + sendJson(socket, { + type: 'reply', + v: 1, + id: message.id, + ok: true, + data: { + agent_id: 'a_relayflow_live_child', + token: 'at_relayflow_live_child', + name: AGENT_NAME, + delivery_ack_seq: 0, + }, + }); + if (message.type === 'agent.deregister') + sendJson(socket, { type: 'reply', v: 1, id: message.id, ok: true, data: {} }); + if (message.type === 'node.register' && !actionSent) { + actionSent = true; + spawnRequests += 1; + sendJson(socket, { + type: 'action.invoke', + v: 1, + invocation_id: INVOCATION_ID, + action: 'spawn', + agent_name: AGENT_NAME, + input: { + name: AGENT_NAME, + cli: 'claude', + channels: [], + harnessConfig: { + runtime: 'native', + command: '/bin/sh', + args: ['-c', childExitCommand(childExitProofPath, childProofNonce, childSessionRef)], + sessionId: childSessionRef, + }, + }, + }); + } + } + }); + socket.once('close', () => sockets.delete(socket)); + socket.once('error', () => sockets.delete(socket)); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const { port } = server.address(); + return { + baseUrl: `http://127.0.0.1:${port}`, + observations: () => ({ + spawnRequests, + eventAttempts: eventBodies.length, + eventBodies, + actionSent, + deliveryAvailable, + controlMessages, + actionResults, + heldResponseCount: heldResponses.size, + }), + successfulSpawnActionResult: () => + actionResults.some( + (result) => + result.invocation_id === INVOCATION_ID && + result.output?.spawned === true && + result.output?.name === AGENT_NAME && + result.error === undefined + ), + expectedChildLifecycle: () => { + const registrationIndex = controlMessages.findIndex( + (message) => + message.type === 'agent.register' && + message.name === AGENT_NAME && + message.invocation_id === INVOCATION_ID && + message.session_ref === childSessionRef + ); + if (registrationIndex < 0) return false; + const lifecycle = controlMessages.slice(registrationIndex + 1); + const activeInventoryIndex = lifecycle.findIndex( + (message) => + message.type === 'inventory.sync' && + message.agents?.some( + (agent) => + agent.name === AGENT_NAME && + agent.invocation_id === INVOCATION_ID && + agent.session_ref === childSessionRef + ) + ); + return ( + activeInventoryIndex >= 0 && + lifecycle + .slice(activeInventoryIndex + 1) + .some( + (message) => + message.type === 'inventory.sync' && !message.agents?.some((agent) => agent.name === AGENT_NAME) + ) + ); + }, + enableDelivery: async () => { + deliveryAvailable = true; + destroyHeldResponses(heldResponses); + }, + close: async () => { + destroyHeldResponses(heldResponses); + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => server.close(resolve)); + }, + }; +} +function childExitCommand(markerPath, nonce, sessionRef) { + return `trap 'status=$?; printf "{\\"nonce\\":\\"${nonce}\\",\\"invocation_id\\":\\"${INVOCATION_ID}\\",\\"session_ref\\":\\"${sessionRef}\\",\\"status\\":%s,\\"pid\\":%s}\\n" "$status" "$$" > "${markerPath}"' 0; sleep 2; exit 23`; +} +function destroyHeldResponses(responses) { + for (const response of responses) { + if (!response.destroyed) response.destroy(); + } + responses.clear(); +} +function sendJson(socket, value) { + socket.write(serverFrame(1, Buffer.from(JSON.stringify(value)))); +} +function serverFrame(opcode, payload) { + const length = payload.length; + if (length < 126) return Buffer.concat([Buffer.from([0x80 | opcode, length]), payload]); + if (length <= 0xffff) { + const header = Buffer.alloc(4); + header[0] = 0x80 | opcode; + header[1] = 126; + header.writeUInt16BE(length, 2); + return Buffer.concat([header, payload]); + } + throw new Error('RelayFlow fake frame unexpectedly exceeds 64KiB.'); +} +function takeFrame(buffer) { + if (buffer.length < 2) return null; + const masked = Boolean(buffer[1] & 0x80); + let length = buffer[1] & 0x7f; + let offset = 2; + if (length === 126) { + if (buffer.length < 4) return null; + length = buffer.readUInt16BE(2); + offset = 4; + } + if (length === 127 || !masked || buffer.length < offset + 4 + length) return null; + const key = buffer.subarray(offset, offset + 4); + const payload = Buffer.from(buffer.subarray(offset + 4, offset + 4 + length)); + for (let index = 0; index < payload.length; index += 1) payload[index] ^= key[index % 4]; + return { opcode: buffer[0] & 0x0f, payload, consumed: offset + 4 + length }; +} +function requestBody(request) { + return new Promise((resolve) => { + const chunks = []; + request.on('data', (chunk) => chunks.push(chunk)); + request.once('end', () => resolve(Buffer.concat(chunks).toString('utf8'))); + }); +} +function json(response, status, value) { + response.writeHead(status, { 'content-type': 'application/json' }); + response.end(JSON.stringify(value)); +} +async function waitFor(check, timeoutMs, message) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await check()) return; + await sleep(25); + } + throw new Error(message); +} +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} +function diagnostic(message, extra) { + return new Error( + `${message} ${JSON.stringify(extra, (_, value) => (typeof value === 'function' ? value() : value)).slice(-12_000)}` + ); +} +function requiredValue(name) { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`Missing ${name}.`); + return value; +} +function requiredDirectory(name) { + return path.resolve(requiredValue(name)); +} +async function requiredExecutable(name) { + const value = path.resolve(requiredValue(name)); + await access(value, fsConstants.X_OK); + return value; +} +function isWithin(root, candidate) { + const relative = path.relative(path.resolve(root), path.resolve(candidate)); + return relative && !relative.startsWith(`..${path.sep}`) && relative !== '..' && !path.isAbsolute(relative); +}