From f9d74428ba5e19cd5ccab8de948264d4b68d8e8a Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 09:28:21 +0200 Subject: [PATCH 1/9] fix(broker): bound an unacknowledged delivery and dead-letter it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `retry_pending_delivery` had no terminal condition for a delivery whose handoff write keeps succeeding and which is never acknowledged. The retry cap gates on `failed_attempts` — consecutive handoff failures — and every successful write resets that counter to zero. `attempts` is cumulative but was bounded by nothing, so such a delivery retried forever, never emitted `message_delivery_failed`, and never reached the dead-letter store. That is the shape of "an agent that goes idle stops receiving": invisible from outside the broker. Add a wall-clock acknowledgement deadline per delivery (30 minutes, `AGENT_RELAY_DELIVERY_MAX_AGE_MS`), plus a cumulative attempt ceiling as a clock-skew backstop. Both drive the identical terminal path the existing failure cases take: `message_delivery_failed` plus a dead-letter entry that `node deadletters` can requeue. The existing `failed_attempts` cap is unchanged. Refs relay#1686 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014CiShnEAJ5Trb7JovG8SdG Session-Id: 2ceeba77-2877-4b1e-b973-bf0453a2e37b --- CHANGELOG.md | 5 + crates/broker/src/runtime/api.rs | 8 + crates/broker/src/runtime/dead_letter.rs | 6 + crates/broker/src/runtime/delivery.rs | 117 +++++++- crates/broker/src/runtime/mod.rs | 37 +++ crates/broker/src/runtime/tests.rs | 348 ++++++++++++++++++++++- crates/broker/src/runtime/util.rs | 20 ++ 7 files changed, 536 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cd719125de..3ca415bc0e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased - Minor] +### Added + +- `AGENT_RELAY_DELIVERY_MAX_AGE_MS` sets how long the broker keeps retrying a message before dead-lettering it (default 30 minutes), and `/api/pending` reports each delivery's `expires_at_ms` / `expires_in_ms`. + ### Changed - `agent-relay fleet spawn --sandbox` now requests Cloud's long-running workload profile and reports the provider Cloud actually selected, enabling Agent37 placement without a provider flag. @@ -17,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A fleet message the broker cannot deliver to its worker is no longer reported back as handled, so it stays outstanding and can be redelivered. - Fleet deliveries the broker rejects are now logged with a reason and sequence number, so a worker that stops receiving messages can be diagnosed from the broker log. - PTY workers no longer exit when Claude Code's folder-trust dialog appears. Relay selects the affirmative option by its label, so both menu orderings work. +- A message the broker hands to an agent that never acknowledges it no longer retries forever in silence. After 30 minutes it reports `message_delivery_failed` and moves to the dead-letter store, where `node deadletters` can requeue it. ## [11.10.3] - 2026-09-05 diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index f09afb5fda..9f729a3397 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -1913,6 +1913,14 @@ impl BrokerRuntime { "attempts": pd.attempts, "queued_at_ms": pd.queued_at_ms, "age_ms": unix_timestamp_millis().saturating_sub(pd.queued_at_ms), + // Makes "how much acknowledgement budget is left" + // answerable from outside the broker, which is the + // whole diagnosis for a recipient that accepts + // every write and never acknowledges (relay#1686). + "expires_at_ms": pd.expires_at_ms, + "expires_in_ms": pd + .expires_at_ms + .saturating_sub(unix_timestamp_millis()), "last_error": pd.last_error, }) }) diff --git a/crates/broker/src/runtime/dead_letter.rs b/crates/broker/src/runtime/dead_letter.rs index 0cfd8e6144..c474e4578d 100644 --- a/crates/broker/src/runtime/dead_letter.rs +++ b/crates/broker/src/runtime/dead_letter.rs @@ -226,6 +226,12 @@ pub(crate) fn requeue_dead_letter( failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: entry.queued_at_ms, + // A requeue is an explicit decision to try this message again, so it + // gets a full fresh acknowledgement budget measured from now. Deriving + // the deadline from the retained `queued_at_ms` — kept for provenance, + // and already past the budget by definition for anything dead-lettered + // by that budget — would re-fail it on the very next maintenance tick. + expires_at_ms: delivery_expires_at_ms(unix_timestamp_millis()), last_error: None, // The dead-lettered entry's withheld fleet ack (if any) was already // dropped when it was dead-lettered — see relay#1310. A requeue is a diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index abbe5f9582..9e7aea4b4a 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -11,6 +11,19 @@ pub(crate) struct PendingDelivery { pub(super) failed_attempts: u32, pub(super) next_retry_at: Instant, pub(super) queued_at_ms: u64, + /// Absolute wall-clock instant (unix millis) after which this delivery is + /// terminally failed and dead-lettered even though every handoff write + /// kept succeeding — the bound `failed_attempts` cannot provide because a + /// successful write resets it. See relay#1686 and [`MAX_DELIVERY_AGE`]. + /// + /// Stored as an absolute deadline rather than derived from `queued_at_ms` + /// on demand so that a requeued dead letter — which keeps its original + /// `queued_at_ms` for provenance — gets a fresh budget instead of being + /// dead on arrival. `0` therefore means "already expired", not "never + /// expires": a construction site that forgets this field fails closed into + /// the dead-letter path, which is observable and requeueable, rather than + /// open into the unbounded retry loop this field exists to close. + pub(super) expires_at_ms: u64, pub(super) last_error: Option, /// Fleet (engine-facing) `delivery_ack` withheld until the worker confirms /// this specific PTY injection landed — echo-verified, or its bounded @@ -42,6 +55,12 @@ pub(crate) struct PersistedPendingDelivery { pub(super) failed_attempts: u32, #[serde(default)] pub(super) queued_at_ms: u64, + /// See `PendingDelivery::expires_at_ms`. A snapshot written before this + /// field existed deserializes as `0`, which the restore path rebuilds from + /// the persisted `queued_at_ms` — an upgrade must not silently hand every + /// restored delivery an expired deadline, nor an unbounded one. + #[serde(default)] + pub(super) expires_at_ms: u64, #[serde(default)] pub(super) last_error: Option, /// See `PendingDelivery::withheld_fleet_ack`. `#[serde(default)]` so a @@ -148,6 +167,14 @@ pub(crate) fn unix_timestamp_millis() -> u64 { chrono::Utc::now().timestamp_millis().max(0) as u64 } +/// Absolute deadline for a delivery queued at `queued_at_ms`. Saturating, so a +/// corrupt far-future queue time yields `u64::MAX` (never expires by age) and +/// is left to the [`MAX_DELIVERY_ATTEMPTS`] backstop rather than wrapping into +/// an immediate expiry. +pub(crate) fn delivery_expires_at_ms(queued_at_ms: u64) -> u64 { + queued_at_ms.saturating_add(delivery_max_age().as_millis() as u64) +} + /// Pending-delivery map with dirty tracking. Any mutable access (insert, /// remove, retry bookkeeping) marks the store dirty via `DerefMut`, letting /// the event loop persist the snapshot immediately after the mutating event @@ -237,6 +264,7 @@ pub(crate) fn save_pending_deliveries( attempts: pd.attempts, failed_attempts: pd.failed_attempts, queued_at_ms: pd.queued_at_ms, + expires_at_ms: pd.expires_at_ms, last_error: pd.last_error.clone(), withheld_fleet_ack: pd.withheld_fleet_ack.clone(), withheld_fleet_ack_floor: pd.withheld_fleet_ack_floor, @@ -258,6 +286,11 @@ pub(crate) fn load_pending_deliveries(path: &Path) -> HashMap HashMap Option { + let now_ms = unix_timestamp_millis(); + if now_ms >= pending.expires_at_ms { + let age_secs = now_ms.saturating_sub(pending.queued_at_ms) / 1_000; + return Some(format!( + "delivery unacknowledged for {age_secs}s across {} attempt(s): exceeded the {}s acknowledgement deadline", + pending.attempts, + delivery_max_age().as_secs() + )); + } + if pending.attempts >= MAX_DELIVERY_ATTEMPTS { + return Some(format!( + "delivery unacknowledged after {} attempts: exceeded the cumulative attempt ceiling of {MAX_DELIVERY_ATTEMPTS}", + pending.attempts + )); + } + None +} + pub(crate) async fn retry_pending_delivery( delivery_id: &DeliveryId, workers: &mut WorkerRegistry, @@ -955,6 +1032,38 @@ pub(crate) async fn retry_pending_delivery( }); } + // The recipient is present and its writes may well still be succeeding — + // this is the case neither cap above can see. `failed_attempts` gates on + // *consecutive handoff failures* and every successful write resets it to + // zero, so a delivery that is written cleanly and never acknowledged had no + // terminal condition at all: it retried forever, emitted no + // `message_delivery_failed`, and never reached the dead-letter store + // (relay#1686). Bound it here, before spending another write on it, and + // drive the identical terminal path the failure cases take. + // + // Checked from inside the retry, not as a separate sweep over the whole + // pending map, so a delivery the worker has already *confirmed* — held back + // only by the cumulative fleet ACK ordering, and skipped by the maintenance + // due-filter — cannot be dead-lettered. That message landed; failing it + // would drop its withheld ack and have the engine redeliver what the agent + // already read. + if let Some(last_error) = terminal_unacked_reason(&pending) { + let removed = pending_deliveries.remove(delivery_id).unwrap_or(pending); + tracing::warn!( + target = "relay_broker::delivery", + worker = %removed.worker_name, + delivery_id = %removed.delivery.delivery_id, + event_id = %removed.delivery.event_id, + attempts = removed.attempts, + reason = %last_error, + "delivery exceeded its acknowledgement budget; dead-lettering" + ); + return Ok(DeliveryAttemptOutcome::Failed { + pending: Box::new(removed), + last_error, + }); + } + match workers .deliver(&pending.worker_name, pending.delivery.clone()) .await diff --git a/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs index 90068ffa87..d9bb2fabe4 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -58,8 +58,45 @@ use crate::worker::{WorkerEvent, WorkerHandle, WorkerRegistry}; use crate::{broker, listen_api, worker_request}; const DEFAULT_DELIVERY_RETRY_MS: u64 = 1_000; +/// Cap on *consecutive* broker-to-worker handoff failures. A successful write +/// resets the counter it reads, so this bounds "the worker keeps refusing the +/// frame" and nothing else. See [`MAX_DELIVERY_AGE`] for the case where every +/// write succeeds and the acknowledgement never comes. const MAX_DELIVERY_RETRIES: u32 = 10; const WAIT_DELIVERY_ACK_TIMEOUT: Duration = Duration::from_secs(5 * 60); +/// Wall-clock budget for one delivery: how long a message may sit +/// unacknowledged before the broker declares it terminally failed, even though +/// every handoff write kept succeeding. +/// +/// `MAX_DELIVERY_RETRIES` cannot bound this case — it gates on +/// `failed_attempts`, which a successful write resets to zero — so a delivery +/// to a recipient that accepts the write and never acknowledges it retried +/// forever and never reported anything (relay#1686). +/// +/// 30 minutes is chosen against the broker's own idea of a reasonable ack: +/// `WAIT_DELIVERY_ACK_TIMEOUT` is 5 minutes and the steer-mode verification +/// window is 5 seconds. A legitimate slow ACK is an agent mid-turn — the write +/// landed in its PTY queue and is consumed when the turn ends — so the budget +/// has to clear the longest plausible turn, not the ack timeout. 30 minutes is +/// 6x the wait-mode ack timeout and 360x the steer window: a recipient that has +/// swallowed a message for half an hour without a single acknowledgement is +/// indistinguishable from a deaf one, and dead-lettering is recoverable +/// (`node deadletters` can requeue) where retrying forever in silence is not. +/// +/// Overridable per-deployment with `AGENT_RELAY_DELIVERY_MAX_AGE_MS`; see +/// [`delivery_max_age`]. +const MAX_DELIVERY_AGE: Duration = Duration::from_secs(30 * 60); +/// Absolute ceiling on *cumulative* attempts, as a backstop for +/// [`MAX_DELIVERY_AGE`]: the deadline is wall-clock, so a frozen or +/// backwards-stepping system clock could otherwise keep a delivery permanently +/// young. Unlike `failed_attempts`, `attempts` is never reset by a successful +/// write, so this can always be reached. +/// +/// Sized so it never fires first under a working clock: the fastest retry +/// cadence is the 5s steer verification window, and 1000 x 5s = ~83 minutes, +/// comfortably past the 30-minute deadline. In wait mode (5 minute cadence) it +/// is days away. If this is what trips, the clock is broken, not the recipient. +const MAX_DELIVERY_ATTEMPTS: u32 = 1_000; const THREAD_HISTORY_LIMIT: usize = 1_000; #[allow(dead_code)] // only http_api_local_delivery_timeout's default; see its own allow const DEFAULT_HTTP_API_LOCAL_DELIVERY_TIMEOUT_MS: u64 = 3_000; diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index d667190bf9..ac12d6bcbf 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -57,7 +57,7 @@ use super::{ DeadLetterEntry, DeadLetterStore, DeliveryAttemptOutcome, InboundContext, InboundQueueOutcome, ObserverTokenMintError, ObserverTokenMintOutcome, PendingDelivery, PendingDeliveryStore, ProtocolHeadlessProvider, RelayWorkspace, RuntimePaths, TypedThreadMessage, MAX_DEAD_LETTERS, - MAX_DELIVERY_RETRIES, + MAX_DELIVERY_ATTEMPTS, MAX_DELIVERY_RETRIES, }; use crate::dedup::DedupCache; use crate::relaycast::{ @@ -408,6 +408,7 @@ fn pending_delivery(worker_name: &str, delivery_id: &str, event_id: &str) -> Pen failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -1126,6 +1127,7 @@ fn make_pending_delivery(delivery_id: &str, worker: &str) -> PendingDelivery { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2548,6 +2550,7 @@ async fn delivery_retry_fails_promptly_when_recipient_is_gone() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: Some("failed writing frame".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2669,6 +2672,7 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2787,6 +2791,341 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { assert_eq!(entry.attempts, MAX_DELIVERY_RETRIES); } +// relay#1686: the defect this test exists for is a delivery whose handoff +// write keeps *succeeding* and which is never acknowledged. `MAX_DELIVERY_RETRIES` +// gates on `failed_attempts` — consecutive handoff failures — and every +// successful write resets that counter to zero, so it can never fire here. The +// only cumulative counter, `attempts`, was bounded by nothing. Such a delivery +// therefore had no terminal condition at all: it retried forever, never emitted +// `message_delivery_failed`, and never reached the dead-letter store, which is +// why "an agent that goes idle stops receiving" is invisible from outside the +// broker. +// +// A `cat` worker is exactly that recipient: it accepts every frame written to +// its stdin and acknowledges none of them. +#[tokio::test] +async fn unacked_delivery_reaches_terminal_dead_letter_while_writes_keep_succeeding() { + let worker_name = "worker-deaf"; + let mut workers = make_worker_registry_with_worker(worker_name).await; + assert!( + workers.has_worker(worker_name), + "the recipient must stay present — a gone recipient already had a terminal path" + ); + + // Queued just inside its acknowledgement budget, so the opening retries are + // ordinary successful handoffs and only elapsed wall-clock time — never a + // write failure — can make this delivery terminal. + let budget_ms = super::delivery_max_age().as_millis() as u64; + let queued_at_ms = super::unix_timestamp_millis().saturating_sub(budget_ms.saturating_sub(400)); + let mut pending_deliveries = HashMap::from([( + DeliveryId::new("del_deaf"), + PendingDelivery { + worker_name: WorkerName::from(worker_name), + delivery: RelayDelivery { + delivery_id: DeliveryId::new("del_deaf"), + event_id: EventId::new("evt_deaf"), + workspace_id: Some(WorkspaceId::new("ws_demo")), + workspace_alias: Some(WorkspaceAlias::new("Demo")), + from: "orchestrator".to_string(), + target: MessageTarget::new(worker_name), + body: "never acknowledged".to_string(), + thread_id: None, + priority: Some(2), + injection_mode: MessageInjectionMode::Wait, + }, + attempts: 0, + failed_attempts: 0, + next_retry_at: Instant::now(), + queued_at_ms, + expires_at_ms: super::delivery_expires_at_ms(queued_at_ms), + last_error: None, + withheld_fleet_ack: None, + withheld_fleet_ack_floor: None, + }, + )]); + + let mut successful_writes = 0_u32; + let mut final_outcome = None; + // Generously more iterations than the deadline needs. Reaching the end of + // this loop is the pre-fix behaviour: retry forever, report nothing. + for _ in 0..400 { + match retry_pending_delivery( + &DeliveryId::new("del_deaf"), + &mut workers, + &mut pending_deliveries, + Duration::from_millis(1), + ) + .await + .expect("a succeeding write must not surface as an error") + { + DeliveryAttemptOutcome::Attempted { .. } => { + successful_writes += 1; + let pending = pending_deliveries + .get("del_deaf") + .expect("an attempted delivery stays pending until it is acknowledged"); + assert_eq!( + pending.failed_attempts, 0, + "every write succeeded, so the failed_attempts cap can never fire — this \ + is the whole reason the delivery needs a second bound" + ); + assert_eq!( + pending.last_error, None, + "a succeeding write records no error to report later" + ); + } + outcome @ DeliveryAttemptOutcome::Failed { .. } => { + final_outcome = Some(outcome); + break; + } + DeliveryAttemptOutcome::Noop => panic!("a present worker's write should not no-op"), + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + + let outcome = final_outcome.expect( + "a delivery that is written successfully and never acknowledged must still reach a \ + terminal state — otherwise it retries forever and nothing ever reports the failure", + ); + assert!( + successful_writes >= 2, + "the delivery must have been handed over successfully several times before it \ + terminated, proving the bound is not just the existing handoff-failure cap: saw {successful_writes}" + ); + let DeliveryAttemptOutcome::Failed { + pending: ref failed, + ref last_error, + } = outcome + else { + unreachable!("matched Failed above"); + }; + assert_eq!( + failed.failed_attempts, 0, + "not one handoff failed; this delivery was terminated by its acknowledgement budget" + ); + assert!( + last_error.contains("unacknowledged"), + "the terminal reason must name the actual condition so it is diagnosable from the \ + event alone: {last_error}" + ); + assert!( + pending_deliveries.is_empty(), + "a terminally failed delivery is removed from the pending map so it cannot keep spinning" + ); + + // A cap with no dead-letter is half a fix: drive the same terminal path the + // existing failure cases take and prove both halves land. + let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); + let mut dead_letters = DeadLetterStore::default(); + emit_delivery_attempt_outcome( + &sdk_out_tx, + &mut dead_letters, + &DeliveryId::new("del_deaf"), + true, + outcome, + ) + .await + .expect("the terminal outcome should emit"); + + let frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) + .await + .expect("an unacknowledged delivery must report message_delivery_failed") + .expect("sdk_out_tx should remain open"); + assert_eq!(frame.payload["kind"], "message_delivery_failed"); + assert_eq!(frame.payload["name"], worker_name); + assert_eq!(frame.payload["delivery_id"], "del_deaf"); + assert_eq!(frame.payload["event_id"], "evt_deaf"); + assert!(frame.payload["lastError"] + .as_str() + .unwrap_or_default() + .contains("unacknowledged")); + + let dead_frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) + .await + .expect("the terminal failure must also emit dead_letter_added") + .expect("sdk_out_tx should remain open"); + assert_eq!(dead_frame.payload["kind"], "dead_letter_added"); + assert_eq!(dead_frame.payload["delivery_id"], "del_deaf"); + + let entry = dead_letters + .get("del_deaf") + .expect("the unacknowledged delivery must land in the dead-letter store, not vanish"); + assert_eq!(entry.delivery.body, "never acknowledged"); + assert_eq!(entry.delivery.event_id, EventId::new("evt_deaf")); + assert!( + entry.reason.contains("unacknowledged"), + "the dead-letter reason must distinguish a never-acknowledged delivery from a failed \ + handoff: {}", + entry.reason + ); + assert_eq!( + dead_letters.len(), + 1, + "the message is retained for requeue rather than silently dropped" + ); +} + +// relay#1686 backstop: the deadline above is wall-clock, so a frozen or +// backwards-stepping system clock could keep an unacknowledged delivery +// permanently "young". `attempts` is the one counter no successful write +// resets, so a cumulative ceiling on it is always reachable. It is sized never +// to fire first under a working clock — if it fires, the clock is broken, not +// the recipient — but it must still be terminal and still dead-letter. +#[tokio::test] +async fn unacked_delivery_terminates_on_the_cumulative_attempt_ceiling() { + let worker_name = "worker-frozen-clock"; + let mut workers = make_worker_registry_with_worker(worker_name).await; + let queued_at_ms = super::unix_timestamp_millis(); + let mut pending_deliveries = HashMap::from([( + DeliveryId::new("del_ceiling"), + PendingDelivery { + worker_name: WorkerName::from(worker_name), + delivery: RelayDelivery { + delivery_id: DeliveryId::new("del_ceiling"), + event_id: EventId::new("evt_ceiling"), + workspace_id: Some(WorkspaceId::new("ws_demo")), + workspace_alias: None, + from: "orchestrator".to_string(), + target: MessageTarget::new(worker_name), + body: "stopped clock".to_string(), + thread_id: None, + priority: None, + injection_mode: MessageInjectionMode::Steer, + }, + attempts: MAX_DELIVERY_ATTEMPTS, + // Deadline unreachable: only the attempt ceiling can end this. + failed_attempts: 0, + next_retry_at: Instant::now(), + queued_at_ms, + expires_at_ms: u64::MAX, + last_error: None, + withheld_fleet_ack: None, + withheld_fleet_ack_floor: None, + }, + )]); + + let outcome = retry_pending_delivery( + &DeliveryId::new("del_ceiling"), + &mut workers, + &mut pending_deliveries, + Duration::from_millis(1), + ) + .await + .expect("the ceiling check must not error"); + + let DeliveryAttemptOutcome::Failed { + pending: ref failed, + ref last_error, + } = outcome + else { + panic!("a delivery at the cumulative attempt ceiling must be terminal: {outcome:?}"); + }; + assert_eq!(failed.failed_attempts, 0, "no handoff ever failed"); + assert!( + last_error.contains("attempt ceiling"), + "the reason must name the ceiling so a broken clock is diagnosable: {last_error}" + ); + assert!(pending_deliveries.is_empty()); + + let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); + let mut dead_letters = DeadLetterStore::default(); + emit_delivery_attempt_outcome( + &sdk_out_tx, + &mut dead_letters, + &DeliveryId::new("del_ceiling"), + true, + outcome, + ) + .await + .expect("the terminal outcome should emit"); + let frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) + .await + .expect("ceiling breach must report message_delivery_failed") + .expect("sdk_out_tx should remain open"); + assert_eq!(frame.payload["kind"], "message_delivery_failed"); + assert_eq!( + dead_letters.len(), + 1, + "the ceiling path is the same terminal path, dead-letter included" + ); +} + +// relay#1686: a requeued dead letter keeps its original `queued_at_ms` for +// provenance, and anything dead-lettered *by* the acknowledgement budget is by +// definition already past it. Deriving the deadline from the queue time would +// therefore re-fail every requeue on the next maintenance tick — an operator +// pressing "redeliver" would watch it die instantly. The requeue gets a fresh +// budget instead. +#[test] +fn requeued_dead_letter_gets_a_fresh_acknowledgement_budget() { + let mut dead_letters = DeadLetterStore::default(); + let mut pending = make_pending_delivery("del_stale", "worker-a"); + // Queued a day ago and dead-lettered for exactly that reason. + pending.queued_at_ms = super::unix_timestamp_millis().saturating_sub(24 * 60 * 60 * 1_000); + pending.expires_at_ms = super::delivery_expires_at_ms(pending.queued_at_ms); + dead_letters.push(DeadLetterEntry::from_pending( + &pending, + "delivery unacknowledged for 86400s", + )); + + let mut pending_deliveries: HashMap = HashMap::new(); + let requeued = super::requeue_dead_letter(&mut dead_letters, &mut pending_deliveries, "del_stale") + .expect("the dead letter should requeue"); + + let now_ms = super::unix_timestamp_millis(); + assert!( + requeued.expires_at_ms > now_ms, + "a requeued delivery must not arrive already expired: expires_at_ms {} vs now {now_ms}", + requeued.expires_at_ms + ); + assert!( + requeued.queued_at_ms < now_ms.saturating_sub(60_000), + "the original queue time is retained for provenance" + ); + assert_eq!(requeued.attempts, 0); +} + +// relay#1686: a snapshot written by a broker that predates the deadline field +// must neither load unbounded (the defect) nor load already expired (an +// upgrade that dead-letters everything in flight). It is rebuilt from the +// persisted queue time, so a delivery a previous broker had been retrying for +// hours does not get its clock reset by the restart either. +#[test] +fn legacy_pending_delivery_snapshot_rebuilds_its_acknowledgement_deadline() { + let dir = tempfile::tempdir().expect("tempdir should create"); + let path = dir.path().join("pending-deliveries.json"); + let mut delivery = make_pending_delivery("del_legacy_deadline", "worker-a"); + let queued_at_ms = super::unix_timestamp_millis().saturating_sub(30_000); + delivery.queued_at_ms = queued_at_ms; + let deliveries = HashMap::from([(DeliveryId::new("del_legacy_deadline"), delivery)]); + super::save_pending_deliveries(&path, &deliveries).expect("pending delivery should save"); + let mut json: Value = serde_json::from_slice( + &std::fs::read(&path).expect("pending delivery snapshot should read"), + ) + .expect("pending delivery snapshot should parse"); + json[0] + .as_object_mut() + .expect("pending delivery entry should be an object") + .remove("expires_at_ms"); + std::fs::write( + &path, + serde_json::to_vec(&json).expect("legacy snapshot encodes"), + ) + .expect("legacy pending snapshot should write"); + + let loaded = load_pending_deliveries(&path); + let restored = &loaded["del_legacy_deadline"]; + assert_eq!( + restored.expires_at_ms, + super::delivery_expires_at_ms(queued_at_ms), + "a pre-relay#1686 snapshot must come back with a deadline measured from when the \ + message was queued — not unbounded, and not reset by the restart" + ); + assert!( + restored.expires_at_ms > super::unix_timestamp_millis(), + "a delivery still inside its budget must not be dead-lettered by the upgrade itself" + ); +} + #[tokio::test] async fn delivery_retry_success_clears_stale_last_error() { let worker_name = "worker-clear-error"; @@ -2811,6 +3150,7 @@ async fn delivery_retry_success_clears_stale_last_error() { failed_attempts: 1, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: Some("old transient failure".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3739,6 +4079,7 @@ fn drop_pending_for_worker_removes_only_matching_entries() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3764,6 +4105,7 @@ fn drop_pending_for_worker_removes_only_matching_entries() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3796,6 +4138,7 @@ async fn dropped_pending_deliveries_emit_terminal_message_failures() { failed_attempts: 1, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: Some("previous blip".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3868,6 +4211,7 @@ fn should_clear_pending_delivery_when_event_id_matches() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3905,6 +4249,7 @@ fn clear_pending_delivery_returns_none_for_stale_event_id() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4324,6 +4669,7 @@ fn should_clear_pending_delivery_without_event_id_for_compatibility() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), + expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, diff --git a/crates/broker/src/runtime/util.rs b/crates/broker/src/runtime/util.rs index 35fcae5787..2cc28d5556 100644 --- a/crates/broker/src/runtime/util.rs +++ b/crates/broker/src/runtime/util.rs @@ -237,6 +237,26 @@ pub(crate) fn delivery_retry_interval() -> Duration { Duration::from_millis(ms.max(50)) } +/// Wall-clock budget a single delivery gets before it is terminally failed and +/// dead-lettered, defaulting to [`MAX_DELIVERY_AGE`]. +/// +/// The floor is the steer-mode verification window: a budget shorter than the +/// window the broker itself waits for an echo confirmation would dead-letter +/// deliveries that are still normally in flight. There is deliberately no +/// value that disables the deadline — an unbounded delivery is the defect this +/// budget exists to close (relay#1686). +pub(crate) fn delivery_max_age() -> Duration { + let configured = std::env::var("AGENT_RELAY_DELIVERY_MAX_AGE_MS") + .ok() + .and_then(|raw| raw.trim().parse::().ok()) + .map(Duration::from_millis) + .unwrap_or(MAX_DELIVERY_AGE); + std::cmp::max( + configured, + crate::broker::delivery_verification::VERIFICATION_WINDOW, + ) +} + // No longer called from production code — the HTTP/sidecar send path // (runtime/api.rs) no longer attempts direct local delivery, so there's // nothing left to bound with a "local delivery" timeout. Kept (with its From 209a3b796372fc7e42c30b8482b6f7761820346a Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 10:25:10 +0200 Subject: [PATCH 2/9] test(pr-proof): prove an unacknowledged delivery becomes terminal Sweep the acknowledgement deadline on the maintenance tick rather than only on the next scheduled retry, so a `Wait`-mode delivery is not held up to five extra minutes past its budget. `retry_pending_delivery` still owns the decision; this only decides when it is asked. Add the RelayFlow proof case. The recipient is a real PTY worker whose child never reads stdin: a body larger than the tty input queue wedges the injection write, so no `delivery_injected` and no acknowledgement ever follow, while every retry's handoff to the worker keeps succeeding. Base retries for a minute with attempts climbing and nothing terminal; head dead-letters at its deadline with a reason naming the condition. Refs relay#1686 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014CiShnEAJ5Trb7JovG8SdG Session-Id: 2ceeba77-2877-4b1e-b973-bf0453a2e37b --- crates/broker/src/runtime/maintenance.rs | 10 +- .../case.json | 21 + .../run.mjs | 431 ++++++++++++++++++ 3 files changed, 461 insertions(+), 1 deletion(-) create mode 100644 tests/relayflows/cases/1686-unacked-delivery-retries-forever/case.json create mode 100644 tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 2d69f15476..170a946659 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -149,6 +149,13 @@ impl BrokerRuntime { ); } + // A delivery past its acknowledgement deadline is picked up here even + // when its next retry is not due yet. Without this the deadline would + // only be noticed on the next scheduled retry, which in `Wait` mode is + // up to `WAIT_DELIVERY_ACK_TIMEOUT` (5 minutes) away — the bound would + // hold, but late. `retry_pending_delivery` still owns the decision; + // this only decides when it gets asked. See relay#1686. + let now_ms = unix_timestamp_millis(); let due_ids: Vec = pending_deliveries .iter() .filter_map(|(delivery_id, pending)| { @@ -156,7 +163,8 @@ impl BrokerRuntime { pending.withheld_fleet_ack.as_ref().is_some_and(|deliver| { fleet_delivery_book.is_delivery_confirmation_held(deliver) }); - if pending.next_retry_at <= now && !confirmation_is_held { + let past_deadline = now_ms >= pending.expires_at_ms; + if (pending.next_retry_at <= now || past_deadline) && !confirmation_is_held { Some(delivery_id.clone()) } else { None diff --git a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/case.json b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/case.json new file mode 100644 index 0000000000..ae05707ab1 --- /dev/null +++ b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/case.json @@ -0,0 +1,21 @@ +{ + "version": 1, + "id": "1686-unacked-delivery-retries-forever", + "kind": "bugfix", + "title": "A delivery the recipient never acknowledges retries forever and is never reported", + "requirements": ["broker-linux-x64"], + "runner": { + "command": ["node", "tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs"] + }, + "timeoutSeconds": 900, + "expected": { + "base": { + "outcome": "bug", + "signature": "unacked_delivery_retries_without_bound" + }, + "head": { + "outcome": "fixed", + "signature": "unacked_delivery_dead_lettered_at_its_deadline" + } + } +} diff --git a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs new file mode 100644 index 0000000000..88bb9a4e69 --- /dev/null +++ b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs @@ -0,0 +1,431 @@ +/** + * relay#1686 — a delivery the recipient never acknowledges retries forever and + * is never reported. + * + * `retry_pending_delivery` had exactly two terminal conditions: + * `failed_attempts >= MAX_DELIVERY_RETRIES` and "recipient gone". The first + * gates on *consecutive handoff failures*, and every successful write resets + * that counter to zero. `attempts` is cumulative but was bounded by nothing. + * So a delivery whose handoff keeps succeeding and which is never acknowledged + * had no terminal state at all: it retried forever, emitted no + * `message_delivery_failed`, and never reached the dead-letter store. From + * outside the broker the whole condition is invisible — which is what makes + * "an agent that goes idle stops receiving" so hard to see. + * + * The recipient here is a real PTY worker whose child never reads its stdin. + * That matters, because a healthy PTY worker cannot produce this shape: after + * the injection write is confirmed, the broker's own pty_worker acknowledges + * the delivery either on echo verification or on its timeout fallback. The + * acknowledgement is withheld only while the *write itself* has not completed — + * "a wedged drainer blocks in this arm instead of emitting a false + * `delivery_injected`" (pty_worker.rs). A child that never reads leaves the tty + * input queue full after a few kilobytes, so a body far larger than that queue + * wedges the drainer for the life of the case. Meanwhile every retry's handoff + * — a send on the worker's command channel, not the tty write — keeps + * succeeding, which is precisely the counter reset that disarms the existing + * cap. + * + * Observed through `GET /api/status` (pending deliveries, with their attempt + * counts) and `GET /api/dead-letters`, both of which exist on either arm. + * + * Base: after a full minute the delivery is still pending, its attempts have + * climbed, and the dead-letter store is empty. Nothing terminal, nothing + * reported. + * Head: the delivery leaves the pending map at its acknowledgement deadline and + * lands in the dead-letter store with a reason naming the condition, and + * `message_delivery_failed` is what put it there. + * + * The deadline is set to `DEADLINE_MS` through `AGENT_RELAY_DELIVERY_MAX_AGE_MS` + * so the case does not have to sit through the 30-minute default. Both arms get + * the identical environment; the base broker has no such variable and ignores + * it, which is itself part of what is being shown. + * + * Control: both arms must first observe the delivery pending with at least one + * attempt recorded. Without that, a head result of "not pending any more" could + * equally mean the message was delivered normally, and a base result of "no + * dead letters" could mean the broker never accepted the message at all. + */ +import { execFileSync, spawn } from 'node:child_process'; +import { chmod, mkdtemp, mkdir, readFile, rm, writeFile } from 'node:fs/promises'; +import { createServer } from 'node:net'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import process from 'node:process'; +import { fileURLToPath } from 'node:url'; + +import { ensureEngine, startEngine } from '../1593-parked-agent-orphaned-receipt/relaycast-engine.mjs'; + +const CASE_ID = '1686-unacked-delivery-retries-forever'; +const AGENT = 'deaf-probe'; +const BROKER_API_KEY = 'rk_proof_broker_api_key'; +const READY_TIMEOUT_MS = 90_000; + +/** Acknowledgement budget both arms are configured with. */ +const DEADLINE_MS = 10_000; +/** + * How long the head broker is given to act on that budget. The deadline is + * swept on the broker's 500ms maintenance tick, so this is generous by an order + * of magnitude. + */ +const HEAD_WINDOW_MS = 45_000; +/** + * How long the base broker is watched for any terminal outcome. Six times the + * deadline and four times the head window: if a bound existed anywhere in the + * base broker's retry path, it would have fired well inside this. + */ +const BASE_WINDOW_MS = 60_000; +/** + * Message body size. The tty input queue is a few kilobytes, so a body this + * size cannot be written to a child that never reads, and the injection write + * never completes. + */ +const BODY_BYTES = 96 * 1024; + +const targetDir = requiredValue('RELAY_PR_PROOF_TARGET_DIR'); +const harnessDir = requiredValue('RELAY_PR_PROOF_HARNESS_DIR'); +const binaryPath = requiredValue('RELAY_PR_PROOF_BROKER_BINARY'); +const resultPath = requiredValue('RELAY_PR_PROOF_RESULT_PATH'); +const arm = requiredValue('RELAY_PR_PROOF_ARM'); +if (arm !== 'base' && arm !== 'head') { + throw new Error(`RELAY_PR_PROOF_ARM must be base or head, received ${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}.`); +} +const runnerPath = fileURLToPath(import.meta.url); +if (!isWithin(harnessDir, runnerPath)) { + throw new Error('The RelayFlow runner must execute from the exact-head harness checkout.'); +} + +const workDir = await mkdtemp(path.join(tmpdir(), 'relayflow-1686-')); +const engineDir = path.join(workDir, 'engine'); +const stateDir = path.join(workDir, 'state'); +const binDir = path.join(workDir, 'bin'); +await mkdir(stateDir, { recursive: true }); +await mkdir(binDir, { recursive: true }); + +const diag = []; +const log = (line) => diag.push(`${String(line).trimEnd()}\n`); +let engine; +let broker; + +try { + // The deaf recipient: it announces itself so the worker reaches readiness, + // then never reads a byte of stdin for the rest of the case. + const wedgedCliPath = path.join(binDir, 'wedged'); + await writeFile( + wedgedCliPath, + ['#!/bin/sh', "printf 'RELAYFLOW_1686_READY\\n'", 'while :; do sleep 3600; done', ''].join('\n'), + { encoding: 'utf8', mode: 0o700 } + ); + await chmod(wedgedCliPath, 0o700); + + const serveBin = await ensureEngine(engineDir, log); + const enginePort = await freePort(); + const engineUrl = `http://127.0.0.1:${enginePort}`; + engine = await startEngine(serveBin, engineDir, enginePort, log); + const eng = engineClient(engineUrl); + await waitFor(async () => { + if (engine.exitCode !== null) throw new Error(`engine exited with code ${engine.exitCode}`); + await fetch(engineUrl); + return true; + }, 'the Relaycast engine to accept connections'); + + const ws = await eng('POST', '/v1/workspaces', { name: 'relayflow-1686' }); + const workspaceKey = ws.body?.data?.api_key; + if (!workspaceKey) { + throw new Error(`workspace create failed: ${JSON.stringify(ws.body).slice(0, 300)}`); + } + const wsAuth = { authorization: `Bearer ${workspaceKey}` }; + + const nodeId = `node_relayflow_1686_${Date.now()}`; + const nodeReg = await eng( + 'POST', + '/v1/nodes', + { + node_id: nodeId, + name: 'relayflow-1686-node', + kind: 'ws', + role: 'broker', + capabilities: [], + max_agents: 8, + version: 'relayflow/1686', + }, + wsAuth + ); + const nodeToken = nodeReg.body?.data?.token; + if (!nodeToken) { + throw new Error(`node mint failed: ${JSON.stringify(nodeReg.body).slice(0, 300)}`); + } + + // The broker must not inherit this process's own Relaycast credentials, or it + // authenticates against production instead of the engine under test. + broker = spawn( + binaryPath, + ['init', '--api-port', '0', '--api-bind', '127.0.0.1', '--state-dir', stateDir], + { + cwd: workDir, + env: { + PATH: `${binDir}:${process.env.PATH}`, + HOME: workDir, + TMPDIR: process.env.TMPDIR ?? '/tmp', + RELAY_BASE_URL: engineUrl, + RELAYCAST_BASE_URL: engineUrl, + RELAY_API_KEY: workspaceKey, + RELAY_WORKSPACE_KEY: workspaceKey, + RELAY_NODE_TOKEN: nodeToken, + RELAY_NODE_ID: nodeId, + RELAY_BROKER_API_KEY: BROKER_API_KEY, + RELAY_SKIP_TELEMETRY: '1', + // No injection pacing: the whole body is offered to the tty in one + // write, so the wedge is immediate rather than drip-fed. + RELAY_INJECT_RATE_MS: '0', + // The head broker's acknowledgement budget. The base broker has no such + // setting and ignores it. + AGENT_RELAY_DELIVERY_MAX_AGE_MS: String(DEADLINE_MS), + RUST_LOG: 'info', + }, + stdio: ['ignore', 'pipe', 'pipe'], + } + ); + broker.stdout.on('data', (d) => log(`[broker] ${d}`)); + broker.stderr.on('data', (d) => log(`[broker] ${d}`)); + + // The broker publishes its bound port in connection.json — a contract, unlike + // its log output. + const brokerUrl = await waitFor(async () => { + if (broker.exitCode !== null) { + throw new Error(`broker exited early with code ${broker.exitCode}`); + } + const connection = JSON.parse(await readFile(path.join(stateDir, 'connection.json'), 'utf8')); + const url = new URL(connection.url); + if (url.hostname !== '127.0.0.1' || !Number(url.port)) { + throw new Error(`bad connection url ${connection.url}`); + } + return connection.url; + }, 'the broker connection file to publish its bound API port'); + const api = brokerClient(brokerUrl); + await waitFor(() => api('GET', '/api/status').then(() => true), 'the broker API to answer'); + + await api('POST', '/api/spawn', { + name: AGENT, + cli: 'wedged', + transport: 'pty', + skip_relay_prompt: true, + }); + await waitFor(async () => { + const listed = await eng('GET', '/v1/agents', undefined, wsAuth); + return (listed.body?.data ?? []).some((agent) => agent.name === AGENT); + }, 'the agent to register with the real engine'); + + // A real DM through the engine, large enough that it cannot be written to a + // child that never reads. + const sender = await eng('POST', '/v1/agents', { name: 'proof-sender', type: 'agent' }, wsAuth); + const senderToken = sender.body?.data?.token; + if (!senderToken) { + throw new Error(`sender create failed: ${JSON.stringify(sender.body).slice(0, 300)}`); + } + const body = `relay-1686 unacked probe ${'x'.repeat(BODY_BYTES)}`; + const dm = await eng( + 'POST', + '/v1/dm', + { to: AGENT, text: body }, + { authorization: `Bearer ${senderToken}` } + ); + if (dm.status >= 300) { + throw new Error(`DM failed: ${dm.status} ${JSON.stringify(dm.body).slice(0, 300)}`); + } + + // Control, on both arms: the broker accepted the message as a retryable + // delivery and has attempted it. Everything below reads as "no bound" or "a + // bound fired" only because this held first. + const tracked = await waitFor(async () => { + const entry = await pendingEntry(api); + return entry && Number(entry.attempts) >= 1 ? entry : null; + }, 'the DM to become a pending delivery with a recorded attempt'); + const deliveryId = String(tracked.delivery_id); + log(`delivery ${deliveryId} pending with ${tracked.attempts} attempt(s)`); + + const startedAt = Date.now(); + const window = arm === 'head' ? HEAD_WINDOW_MS : BASE_WINDOW_MS; + let terminal = null; + let lastAttempts = Number(tracked.attempts); + while (Date.now() - startedAt < window) { + const dead = await deadLetter(api, deliveryId); + if (dead) { + terminal = { dead, elapsedMs: Date.now() - startedAt }; + break; + } + const entry = await pendingEntry(api); + if (entry) { + lastAttempts = Number(entry.attempts); + } else { + // Gone from pending with nothing in the dead-letter store: either it was + // delivered normally (the premise did not hold) or it was dropped + // silently. Neither is an outcome this case may report. + throw new Error( + `Delivery ${deliveryId} left the pending map without a dead letter after ${ + Date.now() - startedAt + }ms — the wedged-recipient premise did not hold, or the delivery was dropped silently.` + ); + } + await sleep(1_000); + } + + let outcome; + let signature; + let details; + if (!terminal) { + if (arm === 'head') { + throw new Error( + `The head broker left delivery ${deliveryId} pending for ${window}ms with ${lastAttempts} attempts and no dead letter, despite a ${DEADLINE_MS}ms acknowledgement budget.` + ); + } + if (lastAttempts < 1) { + throw new Error(`Delivery ${deliveryId} recorded no attempts; the retry path never ran.`); + } + outcome = 'bug'; + signature = 'unacked_delivery_retries_without_bound'; + details = `The base broker retried delivery ${deliveryId} for ${window}ms to a recipient that acknowledged nothing, reaching ${lastAttempts} attempts. It never became terminal: no message_delivery_failed, no dead letter. Every handoff write succeeded, which resets failed_attempts to zero, so the retry cap can never fire and the cumulative attempts are bounded by nothing.`; + } else { + if (arm === 'base') { + throw new Error( + `The base broker dead-lettered ${deliveryId} after ${terminal.elapsedMs}ms: ${JSON.stringify( + terminal.dead + ).slice(0, 400)}. The defect under test is that it never becomes terminal.` + ); + } + const reason = String(terminal.dead.reason ?? ''); + if (!reason.includes('unacknowledged')) { + throw new Error( + `The head broker dead-lettered ${deliveryId} for the wrong reason (${JSON.stringify( + reason + )}). A cap that reports a handoff failure would misdiagnose a recipient that accepted every write.` + ); + } + if (await pendingEntry(api)) { + throw new Error(`Delivery ${deliveryId} is dead-lettered and still pending; it can still spin.`); + } + outcome = 'fixed'; + signature = 'unacked_delivery_dead_lettered_at_its_deadline'; + details = `The head broker made delivery ${deliveryId} terminal ${terminal.elapsedMs}ms after it went unacknowledged, against a ${DEADLINE_MS}ms budget. It is out of the pending map and in the dead-letter store with reason ${JSON.stringify( + reason + )}, where node deadletters can requeue it — rather than retrying forever with nothing reported.`; + } + + await mkdir(path.dirname(resultPath), { recursive: true }); + await writeFile( + resultPath, + `${JSON.stringify({ version: 1, caseId: CASE_ID, arm, outcome, signature, details })}\n`, + 'utf8' + ); + process.stdout.write(`${signature}\n`); +} catch (error) { + process.stderr.write(`${diag.join('').slice(-12_000)}\n`); + throw error; +} finally { + for (const child of [broker, engine]) await stop(child); + await rm(workDir, { recursive: true, force: true }); +} + +function requiredValue(name) { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`Missing required environment variable ${name}.`); + return value; +} +function isWithin(root, candidate) { + const rel = path.relative(path.resolve(root), path.resolve(candidate)); + return rel !== '' && !rel.startsWith('..') && !path.isAbsolute(rel); +} +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} +function freePort() { + return new Promise((resolve, reject) => { + const probe = createServer(); + probe.unref(); + probe.on('error', reject); + probe.listen(0, '127.0.0.1', () => { + const { port } = probe.address(); + probe.close(() => resolve(port)); + }); + }); +} +function engineClient(baseUrl) { + return async (method, route, body, headers = {}) => { + const res = await fetch(`${baseUrl}${route}`, { + method, + headers: { 'content-type': 'application/json', ...headers }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + }); + const text = await res.text(); + let parsed = {}; + try { + parsed = text ? JSON.parse(text) : {}; + } catch { + parsed = { raw: text }; + } + return { status: res.status, body: parsed }; + }; +} +function brokerClient(baseUrl) { + return async (method, route, body) => { + const res = await fetch(`${baseUrl}${route}`, { + method, + headers: { 'content-type': 'application/json', 'x-api-key': BROKER_API_KEY }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + }); + const text = await res.text(); + let parsed = {}; + try { + parsed = text ? JSON.parse(text) : {}; + } catch { + parsed = { raw: text }; + } + if (!res.ok) throw new Error(`${method} ${route} -> ${res.status} ${text.slice(0, 300)}`); + return parsed; + }; +} +async function pendingEntry(api) { + const status = await api('GET', '/api/status'); + const pending = Array.isArray(status.pending) ? status.pending : []; + return pending.find((entry) => entry.worker_name === AGENT) ?? null; +} +async function deadLetter(api, deliveryId) { + const body = await api('GET', '/api/dead-letters'); + const entries = Array.isArray(body.dead_letters) + ? body.dead_letters + : Array.isArray(body.entries) + ? body.entries + : []; + return ( + entries.find((entry) => (entry.delivery_id ?? entry.delivery?.delivery_id) === deliveryId) ?? + null + ); +} +async function waitFor(predicate, label, timeoutMs = READY_TIMEOUT_MS) { + const deadline = Date.now() + timeoutMs; + let last; + while (Date.now() < deadline) { + try { + const value = await predicate(); + if (value) return value; + } catch (error) { + last = error; + } + await sleep(300); + } + throw new Error(`Timed out waiting for ${label}${last ? `: ${last.message}` : ''}.`); +} +async function stop(child) { + if (!child || child.exitCode !== null) return; + child.kill('SIGTERM'); + await Promise.race([new Promise((r) => child.once('exit', r)), sleep(5_000)]); + if (child.exitCode === null) child.kill('SIGKILL'); +} From b0db16b7cee8f7d776ea783a3002f2a59393a3d5 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 11:22:34 +0200 Subject: [PATCH 3/9] style: rustfmt and prettier the relay#1686 changes Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014CiShnEAJ5Trb7JovG8SdG Session-Id: 2ceeba77-2877-4b1e-b973-bf0453a2e37b --- crates/broker/src/runtime/tests.rs | 11 ++++++++--- .../1686-unacked-delivery-retries-forever/run.mjs | 5 +---- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index ac12d6bcbf..5b0a438056 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -3012,12 +3012,16 @@ async fn unacked_delivery_terminates_on_the_cumulative_attempt_ceiling() { .await .expect("the ceiling check must not error"); + assert!( + matches!(outcome, DeliveryAttemptOutcome::Failed { .. }), + "a delivery at the cumulative attempt ceiling must be terminal, got {outcome:?}" + ); let DeliveryAttemptOutcome::Failed { pending: ref failed, ref last_error, } = outcome else { - panic!("a delivery at the cumulative attempt ceiling must be terminal: {outcome:?}"); + unreachable!("asserted Failed above"); }; assert_eq!(failed.failed_attempts, 0, "no handoff ever failed"); assert!( @@ -3068,8 +3072,9 @@ fn requeued_dead_letter_gets_a_fresh_acknowledgement_budget() { )); let mut pending_deliveries: HashMap = HashMap::new(); - let requeued = super::requeue_dead_letter(&mut dead_letters, &mut pending_deliveries, "del_stale") - .expect("the dead letter should requeue"); + let requeued = + super::requeue_dead_letter(&mut dead_letters, &mut pending_deliveries, "del_stale") + .expect("the dead letter should requeue"); let now_ms = super::unix_timestamp_millis(); assert!( diff --git a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs index 88bb9a4e69..0128a80c77 100644 --- a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs +++ b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs @@ -404,10 +404,7 @@ async function deadLetter(api, deliveryId) { : Array.isArray(body.entries) ? body.entries : []; - return ( - entries.find((entry) => (entry.delivery_id ?? entry.delivery?.delivery_id) === deliveryId) ?? - null - ); + return entries.find((entry) => (entry.delivery_id ?? entry.delivery?.delivery_id) === deliveryId) ?? null; } async function waitFor(predicate, label, timeoutMs = READY_TIMEOUT_MS) { const deadline = Date.now() + timeoutMs; From 4466f9affc64f895bac69d84949aca27ef133bb5 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 14:46:55 +0200 Subject: [PATCH 4/9] fix(broker): address review on the delivery acknowledgement bound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Read the status endpoint's actual field. `GET /api/status` returns pending entries under `pending_deliveries`; the proof case read `status.pending`, which never exists, so its control timed out and the case proved nothing. This was the cause of the red proof run, not the reproduction itself — a local probe against a base-era broker shows the delivery pending with attempts climbing 1..8 on a ~5.5s cadence, `last_error` null on every one, and the dead-letter store empty throughout. Scale the cumulative attempt ceiling from the configured budget and retry cadence. A fixed 1000 dead-lettered a `Steer` delivery after ~83 minutes of five-second attempts however high `AGENT_RELAY_DELIVERY_MAX_AGE_MS` was set, silently capping the setting an operator asked for and making the backstop — not the deadline — the thing that decided. Clamp the configured age at both ends. `u64::MAX` was accepted as an effectively unbounded budget, reinstating by configuration the exact "retries forever, reports nothing" state this change exists to close. Floor each delivery's budget at its own acknowledgement timeout. A budget below the 5 minute `Wait` timeout dead-lettered wait-mode deliveries before the recipient had one full window to answer. The floor is per-delivery, not global, so steer-only deployments can still configure a short budget. Persist the deadline as `Option`. In memory `0` means "already expired", so a `0` sentinel made the load path hand a genuinely exhausted delivery a fresh budget on every restart; only an absent field means "no deadline was ever recorded". Apply queued worker events before the deadline sweep. `worker_event_rx` and `reap_tick` are sibling `select!` arms, so a `delivery_verified` the worker had already sent but the loop had not yet handled was invisible to the sweep, which would dead-letter a message the agent did receive — dropping its withheld fleet ack and making the engine redeliver it. Also: widen the synthetic margin in the deadline test from 400ms to 2s so CI contention cannot fail it for scheduling reasons; derive the legacy snapshot test's offset from the budget instead of a hard-coded 30s; bound every request in the proof case as sibling cases do; and correct the changelog to `GET /api/status`, impact-first and split per impact. Refs relay#1686 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014CiShnEAJ5Trb7JovG8SdG Session-Id: 2ceeba77-2877-4b1e-b973-bf0453a2e37b --- CHANGELOG.md | 5 +- crates/broker/src/runtime/dead_letter.rs | 17 +- crates/broker/src/runtime/delivery.rs | 128 ++++++++--- crates/broker/src/runtime/maintenance.rs | 28 +++ crates/broker/src/runtime/mod.rs | 38 +++- crates/broker/src/runtime/tests.rs | 202 ++++++++++++++++-- crates/broker/src/runtime/util.rs | 17 +- .../run.mjs | 50 +++-- 8 files changed, 391 insertions(+), 94 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ca415bc0e..f0af8010e2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- `AGENT_RELAY_DELIVERY_MAX_AGE_MS` sets how long the broker keeps retrying a message before dead-lettering it (default 30 minutes), and `/api/pending` reports each delivery's `expires_at_ms` / `expires_in_ms`. +- `node deadletters` now surfaces messages an agent never acknowledged, so a silently deaf recipient is visible and its messages can be requeued. +- `GET /api/status` reports how much acknowledgement budget each pending delivery has left, so a stalled delivery can be spotted before it is dead-lettered. ### Changed @@ -21,7 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A fleet message the broker cannot deliver to its worker is no longer reported back as handled, so it stays outstanding and can be redelivered. - Fleet deliveries the broker rejects are now logged with a reason and sequence number, so a worker that stops receiving messages can be diagnosed from the broker log. - PTY workers no longer exit when Claude Code's folder-trust dialog appears. Relay selects the affirmative option by its label, so both menu orderings work. -- A message the broker hands to an agent that never acknowledges it no longer retries forever in silence. After 30 minutes it reports `message_delivery_failed` and moves to the dead-letter store, where `node deadletters` can requeue it. +- A message the broker hands to an agent that never acknowledges it no longer retries forever in silence. After 30 minutes it reports `message_delivery_failed` and moves to the dead-letter store; `AGENT_RELAY_DELIVERY_MAX_AGE_MS` tunes the budget. ## [11.10.3] - 2026-09-05 diff --git a/crates/broker/src/runtime/dead_letter.rs b/crates/broker/src/runtime/dead_letter.rs index c474e4578d..26e0de05bc 100644 --- a/crates/broker/src/runtime/dead_letter.rs +++ b/crates/broker/src/runtime/dead_letter.rs @@ -219,6 +219,16 @@ pub(crate) fn requeue_dead_letter( // The event id (message identity) is preserved. let mut delivery = entry.delivery; delivery.delivery_id = DeliveryId::new(format!("del_{}", Uuid::new_v4().simple())); + // A requeue is an explicit decision to try this message again, so it gets a + // full fresh acknowledgement budget measured from now. Deriving the + // deadline from the retained `queued_at_ms` — kept for provenance, and + // already past the budget by definition for anything dead-lettered by that + // budget — would re-fail it on the very next maintenance tick. + let expires_at_ms = delivery_expires_at_ms( + unix_timestamp_millis(), + &delivery.injection_mode, + delivery_retry_interval(), + ); let pending = PendingDelivery { worker_name: entry.worker_name, delivery, @@ -226,12 +236,7 @@ pub(crate) fn requeue_dead_letter( failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: entry.queued_at_ms, - // A requeue is an explicit decision to try this message again, so it - // gets a full fresh acknowledgement budget measured from now. Deriving - // the deadline from the retained `queued_at_ms` — kept for provenance, - // and already past the budget by definition for anything dead-lettered - // by that budget — would re-fail it on the very next maintenance tick. - expires_at_ms: delivery_expires_at_ms(unix_timestamp_millis()), + expires_at_ms, last_error: None, // The dead-lettered entry's withheld fleet ack (if any) was already // dropped when it was dead-lettered — see relay#1310. A requeue is a diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index 9e7aea4b4a..d604679b0b 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -55,12 +55,13 @@ pub(crate) struct PersistedPendingDelivery { pub(super) failed_attempts: u32, #[serde(default)] pub(super) queued_at_ms: u64, - /// See `PendingDelivery::expires_at_ms`. A snapshot written before this - /// field existed deserializes as `0`, which the restore path rebuilds from - /// the persisted `queued_at_ms` — an upgrade must not silently hand every - /// restored delivery an expired deadline, nor an unbounded one. + /// See `PendingDelivery::expires_at_ms`. `Option`, not a `0` sentinel: in + /// memory `0` means "already expired", so a snapshot that genuinely holds + /// an expired deadline must round-trip as expired. `None` — the only thing + /// a pre-relay#1686 snapshot can produce, via `#[serde(default)]` — is the + /// distinct "no deadline was ever recorded" case the restore path rebuilds. #[serde(default)] - pub(super) expires_at_ms: u64, + pub(super) expires_at_ms: Option, #[serde(default)] pub(super) last_error: Option, /// See `PendingDelivery::withheld_fleet_ack`. `#[serde(default)]` so a @@ -167,12 +168,70 @@ pub(crate) fn unix_timestamp_millis() -> u64 { chrono::Utc::now().timestamp_millis().max(0) as u64 } +/// Acknowledgement budget for one delivery: the configured wall-clock age, but +/// never shorter than this delivery's own acknowledgement timeout. +/// +/// The floor is per-delivery, not global, because the timeout is per-mode — 5 +/// minutes for `Wait`, the 5 second verification window for `Steer`. A budget +/// below a delivery's own timeout would dead-letter it before the recipient +/// had been given one full window to acknowledge, which would make the bound a +/// statement about the broker's impatience rather than the recipient's +/// silence. Flooring globally at the `Wait` timeout instead would force that 5 +/// minutes onto steer-only deployments that deliberately configured less. +pub(crate) fn delivery_budget( + injection_mode: &MessageInjectionMode, + retry_interval: Duration, +) -> Duration { + std::cmp::max( + delivery_max_age(), + delivery_ack_timeout(injection_mode, retry_interval), + ) +} + /// Absolute deadline for a delivery queued at `queued_at_ms`. Saturating, so a /// corrupt far-future queue time yields `u64::MAX` (never expires by age) and -/// is left to the [`MAX_DELIVERY_ATTEMPTS`] backstop rather than wrapping into -/// an immediate expiry. -pub(crate) fn delivery_expires_at_ms(queued_at_ms: u64) -> u64 { - queued_at_ms.saturating_add(delivery_max_age().as_millis() as u64) +/// is left to the attempt-ceiling backstop rather than wrapping into an +/// immediate expiry. +pub(crate) fn delivery_expires_at_ms( + queued_at_ms: u64, + injection_mode: &MessageInjectionMode, + retry_interval: Duration, +) -> u64 { + queued_at_ms.saturating_add(delivery_budget(injection_mode, retry_interval).as_millis() as u64) +} + +/// Cumulative-attempt ceiling for one delivery. +/// +/// Derived from the configured budget and this delivery's retry cadence rather +/// than fixed, so raising `AGENT_RELAY_DELIVERY_MAX_AGE_MS` cannot be defeated +/// by a constant: a fixed 1000 would dead-letter a `Steer` delivery after +/// ~83 minutes of 5-second attempts even when the operator asked for longer and +/// the clock is healthy. Scaling keeps the ceiling strictly behind the +/// deadline under a working clock — it exists only for a clock that has +/// stopped or stepped backwards — while staying finite, which is the whole +/// point of having it. +#[cfg(test)] +pub(crate) fn delivery_attempt_ceiling_for_test( + injection_mode: &MessageInjectionMode, + retry_interval: Duration, +) -> u32 { + delivery_attempt_ceiling(injection_mode, retry_interval) +} + +fn delivery_attempt_ceiling( + injection_mode: &MessageInjectionMode, + retry_interval: Duration, +) -> u32 { + let budget_ms = delivery_budget(injection_mode, retry_interval).as_millis(); + let cadence_ms = delivery_ack_timeout(injection_mode, retry_interval) + .as_millis() + .max(1); + let projected = budget_ms / cadence_ms; + let scaled = projected.saturating_mul(u128::from(ATTEMPT_CEILING_HEADROOM)); + scaled.clamp( + u128::from(MIN_DELIVERY_ATTEMPT_CEILING), + u128::from(u32::MAX), + ) as u32 } /// Pending-delivery map with dirty tracking. Any mutable access (insert, @@ -264,7 +323,7 @@ pub(crate) fn save_pending_deliveries( attempts: pd.attempts, failed_attempts: pd.failed_attempts, queued_at_ms: pd.queued_at_ms, - expires_at_ms: pd.expires_at_ms, + expires_at_ms: Some(pd.expires_at_ms), last_error: pd.last_error.clone(), withheld_fleet_ack: pd.withheld_fleet_ack.clone(), withheld_fleet_ack_floor: pd.withheld_fleet_ack_floor, @@ -291,6 +350,7 @@ pub(crate) fn load_pending_deliveries(path: &Path) -> HashMap HashMap Option { +fn terminal_unacked_reason(pending: &PendingDelivery, retry_interval: Duration) -> Option { let now_ms = unix_timestamp_millis(); if now_ms >= pending.expires_at_ms { let age_secs = now_ms.saturating_sub(pending.queued_at_ms) / 1_000; + let budget_secs = + delivery_budget(&pending.delivery.injection_mode, retry_interval).as_secs(); return Some(format!( - "delivery unacknowledged for {age_secs}s across {} attempt(s): exceeded the {}s acknowledgement deadline", + "delivery unacknowledged for {age_secs}s across {} attempt(s): exceeded the {budget_secs}s acknowledgement deadline", pending.attempts, - delivery_max_age().as_secs() )); } - if pending.attempts >= MAX_DELIVERY_ATTEMPTS { + let ceiling = delivery_attempt_ceiling(&pending.delivery.injection_mode, retry_interval); + if pending.attempts >= ceiling { return Some(format!( - "delivery unacknowledged after {} attempts: exceeded the cumulative attempt ceiling of {MAX_DELIVERY_ATTEMPTS}", + "delivery unacknowledged after {} attempts: exceeded the cumulative attempt ceiling of {ceiling}", pending.attempts )); } @@ -1047,7 +1117,7 @@ pub(crate) async fn retry_pending_delivery( // due-filter — cannot be dead-lettered. That message landed; failing it // would drop its withheld ack and have the engine redeliver what the agent // already read. - if let Some(last_error) = terminal_unacked_reason(&pending) { + if let Some(last_error) = terminal_unacked_reason(&pending, retry_interval) { let removed = pending_deliveries.remove(delivery_id).unwrap_or(pending); tracing::warn!( target = "relay_broker::delivery", diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 170a946659..d9425faf5a 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -3,7 +3,35 @@ use super::*; use crate::terminal_control::TerminalToCloud; impl BrokerRuntime { + /// Apply worker events that are already queued, without waiting for any. + /// Returns once the channel is momentarily empty, `limit` events have been + /// applied, or the channel is closed (the `select!` arm owns that + /// transition, so it is simply left alone here). + async fn drain_ready_worker_events(&mut self, limit: usize) { + for _ in 0..limit { + match self.worker_event_rx.try_recv() { + Ok(event) => self.handle_worker_event(event).await, + Err(_) => return, + } + } + } + pub(super) async fn handle_maintenance_tick(&mut self) { + // Worker events already sitting in the channel are applied before this + // tick reads any worker state. `worker_event_rx` and `reap_tick` are + // sibling arms of one `select!`, so their order is arbitrary: without + // this, a `delivery_verified` that the worker has already sent but the + // loop has not yet handled would be invisible to the deadline sweep + // below, which would dead-letter a delivery the agent did receive — + // dropping its withheld fleet ack and making the engine redeliver a + // message the agent already read. Draining first makes the sweep read + // the freshest state the broker actually has. See relay#1686. + // + // Bounded so a busy worker cannot starve the rest of the tick; anything + // left is handled by the normal `select!` arm on the next iterations. + self.drain_ready_worker_events(MAX_DRAINED_WORKER_EVENTS_PER_TICK) + .await; + let paths = &self.paths; let state = &mut self.state; let sdk_out_tx = &self.sdk_out_tx; diff --git a/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs index d9bb2fabe4..3f95ad2dc7 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -86,17 +86,35 @@ const WAIT_DELIVERY_ACK_TIMEOUT: Duration = Duration::from_secs(5 * 60); /// Overridable per-deployment with `AGENT_RELAY_DELIVERY_MAX_AGE_MS`; see /// [`delivery_max_age`]. const MAX_DELIVERY_AGE: Duration = Duration::from_secs(30 * 60); -/// Absolute ceiling on *cumulative* attempts, as a backstop for -/// [`MAX_DELIVERY_AGE`]: the deadline is wall-clock, so a frozen or -/// backwards-stepping system clock could otherwise keep a delivery permanently -/// young. Unlike `failed_attempts`, `attempts` is never reset by a successful -/// write, so this can always be reached. +/// Floor for the *cumulative* attempt ceiling, which is otherwise scaled from +/// the configured budget (see `delivery_attempt_ceiling`). /// -/// Sized so it never fires first under a working clock: the fastest retry -/// cadence is the 5s steer verification window, and 1000 x 5s = ~83 minutes, -/// comfortably past the 30-minute deadline. In wait mode (5 minute cadence) it -/// is days away. If this is what trips, the clock is broken, not the recipient. -const MAX_DELIVERY_ATTEMPTS: u32 = 1_000; +/// The ceiling is a backstop for [`MAX_DELIVERY_AGE`]: the deadline is +/// wall-clock, so a frozen or backwards-stepping system clock could otherwise +/// keep a delivery permanently young. Unlike `failed_attempts`, `attempts` is +/// never reset by a successful write, so this can always be reached. +/// +/// It must never fire before the deadline under a working clock, which is why +/// it cannot be a constant: at the default 30-minute budget a fixed 1000 sits +/// comfortably past it, but an operator raising +/// `AGENT_RELAY_DELIVERY_MAX_AGE_MS` beyond ~83 minutes would find `Steer` +/// deliveries dead-lettered at 1000 five-second attempts with the deadline +/// still in the future — silently capping the setting they asked for. +const MIN_DELIVERY_ATTEMPT_CEILING: u32 = 1_000; +/// How far past the attempts a healthy clock would need the ceiling to sit. +/// Retries can also be deferred (a `delivery_queued` frame pushes the next one +/// out without consuming an attempt), so the projection is a lower bound on +/// elapsed time per attempt; the headroom keeps the ceiling behind the +/// deadline anyway. +const ATTEMPT_CEILING_HEADROOM: u32 = 4; +/// Upper bound on `AGENT_RELAY_DELIVERY_MAX_AGE_MS`. Without it, a value like +/// `u64::MAX` would be accepted as an effectively unbounded age and defeat the +/// wall-clock termination guarantee this whole change exists to provide — the +/// same "retries forever, reports nothing" state, reachable by configuration. +const MAX_CONFIGURABLE_DELIVERY_AGE: Duration = Duration::from_secs(7 * 24 * 60 * 60); +/// How many already-queued worker events a maintenance tick applies before the +/// rest of its work. Bounded so a chatty worker cannot starve the tick. +const MAX_DRAINED_WORKER_EVENTS_PER_TICK: usize = 256; const THREAD_HISTORY_LIMIT: usize = 1_000; #[allow(dead_code)] // only http_api_local_delivery_timeout's default; see its own allow const DEFAULT_HTTP_API_LOCAL_DELIVERY_TIMEOUT_MS: u64 = 3_000; diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 5b0a438056..d894f66ec6 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -56,8 +56,9 @@ use super::{ take_pending_for_worker, try_inject_pending_relay_message, AgentRuntime, BrokerRuntime, DeadLetterEntry, DeadLetterStore, DeliveryAttemptOutcome, InboundContext, InboundQueueOutcome, ObserverTokenMintError, ObserverTokenMintOutcome, PendingDelivery, PendingDeliveryStore, - ProtocolHeadlessProvider, RelayWorkspace, RuntimePaths, TypedThreadMessage, MAX_DEAD_LETTERS, - MAX_DELIVERY_ATTEMPTS, MAX_DELIVERY_RETRIES, + ProtocolHeadlessProvider, RelayWorkspace, RuntimePaths, TypedThreadMessage, + MAX_CONFIGURABLE_DELIVERY_AGE, MAX_DEAD_LETTERS, MAX_DELIVERY_RETRIES, + WAIT_DELIVERY_ACK_TIMEOUT, }; use crate::dedup::DedupCache; use crate::relaycast::{ @@ -408,7 +409,11 @@ fn pending_delivery(worker_name: &str, delivery_id: &str, event_id: &str) -> Pen failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -1127,7 +1132,11 @@ fn make_pending_delivery(delivery_id: &str, worker: &str) -> PendingDelivery { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2550,7 +2559,11 @@ async fn delivery_retry_fails_promptly_when_recipient_is_gone() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: Some("failed writing frame".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2672,7 +2685,11 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2815,8 +2832,15 @@ async fn unacked_delivery_reaches_terminal_dead_letter_while_writes_keep_succeed // Queued just inside its acknowledgement budget, so the opening retries are // ordinary successful handoffs and only elapsed wall-clock time — never a // write failure — can make this delivery terminal. - let budget_ms = super::delivery_max_age().as_millis() as u64; - let queued_at_ms = super::unix_timestamp_millis().saturating_sub(budget_ms.saturating_sub(400)); + // 2s, not a few hundred ms: the loop below must fit at least two retry + // calls inside the remaining margin, and a contended CI runner can lose + // several hundred milliseconds between iterations. Too tight a margin fails + // this test for scheduling reasons while the deadline logic is correct. + const REMAINING_BUDGET_MS: u64 = 2_000; + let budget_ms = super::delivery_budget(&MessageInjectionMode::Wait, delivery_retry_interval()) + .as_millis() as u64; + let queued_at_ms = super::unix_timestamp_millis() + .saturating_sub(budget_ms.saturating_sub(REMAINING_BUDGET_MS)); let mut pending_deliveries = HashMap::from([( DeliveryId::new("del_deaf"), PendingDelivery { @@ -2837,7 +2861,11 @@ async fn unacked_delivery_reaches_terminal_dead_letter_while_writes_keep_succeed failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms, - expires_at_ms: super::delivery_expires_at_ms(queued_at_ms), + expires_at_ms: super::delivery_expires_at_ms( + queued_at_ms, + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2848,7 +2876,7 @@ async fn unacked_delivery_reaches_terminal_dead_letter_while_writes_keep_succeed let mut final_outcome = None; // Generously more iterations than the deadline needs. Reaching the end of // this loop is the pre-fix behaviour: retry forever, report nothing. - for _ in 0..400 { + for _ in 0..600 { match retry_pending_delivery( &DeliveryId::new("del_deaf"), &mut workers, @@ -2991,7 +3019,7 @@ async fn unacked_delivery_terminates_on_the_cumulative_attempt_ceiling() { priority: None, injection_mode: MessageInjectionMode::Steer, }, - attempts: MAX_DELIVERY_ATTEMPTS, + attempts: u32::MAX, // Deadline unreachable: only the attempt ceiling can end this. failed_attempts: 0, next_retry_at: Instant::now(), @@ -3053,6 +3081,96 @@ async fn unacked_delivery_terminates_on_the_cumulative_attempt_ceiling() { ); } +// relay#1686 review follow-up: the attempt ceiling is a clock-skew backstop, so +// it must never be what ends a delivery while the wall-clock deadline is still +// in the future. A fixed ceiling could not hold that: at 1000 attempts a `Steer` +// delivery retrying every 5s dies at ~83 minutes, so any operator raising +// `AGENT_RELAY_DELIVERY_MAX_AGE_MS` past that would silently get the ceiling +// instead of the budget they configured. Scaling it from the budget is the fix; +// this pins the invariant for both modes. +#[test] +fn attempt_ceiling_stays_behind_a_raised_deadline() { + let _guard = env_test_lock().lock().expect("env test lock"); + std::env::set_var( + "AGENT_RELAY_DELIVERY_MAX_AGE_MS", + &(6 * 60 * 60 * 1_000).to_string(), + ); + let retry_interval = delivery_retry_interval(); + for mode in [MessageInjectionMode::Steer, MessageInjectionMode::Wait] { + let budget = super::delivery_budget(&mode, retry_interval); + let cadence = super::delivery_ack_timeout(&mode, retry_interval); + // The most attempts a healthy clock could record inside the budget. + let attempts_in_budget = (budget.as_millis() / cadence.as_millis().max(1)) as u128; + let ceiling = u128::from(super::delivery_attempt_ceiling_for_test( + &mode, + retry_interval, + )); + assert!( + ceiling > attempts_in_budget, + "{mode:?}: ceiling {ceiling} must sit past the {attempts_in_budget} attempts the \ + configured {}s budget allows, or it — not the deadline — decides", + budget.as_secs() + ); + } + std::env::remove_var("AGENT_RELAY_DELIVERY_MAX_AGE_MS"); +} + +// relay#1686 review follow-up: `AGENT_RELAY_DELIVERY_MAX_AGE_MS` must not be +// able to switch the bound off. An unclamped `u64::MAX` would be accepted as an +// effectively infinite budget and reinstate, by configuration, the exact +// "retries forever and reports nothing" state this change closes. +#[test] +fn configured_delivery_age_is_clamped_at_both_ends() { + let _guard = env_test_lock().lock().expect("env test lock"); + std::env::set_var("AGENT_RELAY_DELIVERY_MAX_AGE_MS", &u64::MAX.to_string()); + assert_eq!( + super::delivery_max_age(), + MAX_CONFIGURABLE_DELIVERY_AGE, + "an unbounded age must be clamped, not honoured" + ); + + std::env::set_var("AGENT_RELAY_DELIVERY_MAX_AGE_MS", "1"); + assert_eq!( + super::delivery_max_age(), + crate::broker::delivery_verification::VERIFICATION_WINDOW, + "a budget below the echo verification window would kill deliveries still in flight" + ); + + // And a Wait delivery is floored again at its own acknowledgement timeout, + // so a short global budget cannot dead-letter it before the recipient has + // had one full window to answer. + assert_eq!( + super::delivery_budget(&MessageInjectionMode::Wait, delivery_retry_interval()), + WAIT_DELIVERY_ACK_TIMEOUT, + ); + assert_eq!( + super::delivery_budget(&MessageInjectionMode::Steer, delivery_retry_interval()), + crate::broker::delivery_verification::VERIFICATION_WINDOW, + ); + std::env::remove_var("AGENT_RELAY_DELIVERY_MAX_AGE_MS"); +} + +// relay#1686 review follow-up: in memory `expires_at_ms == 0` means "already +// expired", so a snapshot holding that must round-trip as expired rather than +// be mistaken for a pre-upgrade snapshot and handed a fresh budget. Only an +// absent field means "no deadline was ever recorded". +#[test] +fn persisted_zero_deadline_round_trips_as_expired() { + let dir = tempfile::tempdir().expect("tempdir should create"); + let path = dir.path().join("pending-deliveries.json"); + let mut delivery = make_pending_delivery("del_expired", "worker-a"); + delivery.expires_at_ms = 0; + let deliveries = HashMap::from([(DeliveryId::new("del_expired"), delivery)]); + super::save_pending_deliveries(&path, &deliveries).expect("pending delivery should save"); + + let loaded = load_pending_deliveries(&path); + assert_eq!( + loaded["del_expired"].expires_at_ms, 0, + "an explicitly expired deadline must survive the restart as expired — rebuilding it \ + would hand a delivery that already exhausted its budget a fresh one on every restart" + ); +} + // relay#1686: a requeued dead letter keeps its original `queued_at_ms` for // provenance, and anything dead-lettered *by* the acknowledgement budget is by // definition already past it. Deriving the deadline from the queue time would @@ -3065,7 +3183,11 @@ fn requeued_dead_letter_gets_a_fresh_acknowledgement_budget() { let mut pending = make_pending_delivery("del_stale", "worker-a"); // Queued a day ago and dead-lettered for exactly that reason. pending.queued_at_ms = super::unix_timestamp_millis().saturating_sub(24 * 60 * 60 * 1_000); - pending.expires_at_ms = super::delivery_expires_at_ms(pending.queued_at_ms); + pending.expires_at_ms = super::delivery_expires_at_ms( + pending.queued_at_ms, + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ); dead_letters.push(DeadLetterEntry::from_pending( &pending, "delivery unacknowledged for 86400s", @@ -3099,7 +3221,13 @@ fn legacy_pending_delivery_snapshot_rebuilds_its_acknowledgement_deadline() { let dir = tempfile::tempdir().expect("tempdir should create"); let path = dir.path().join("pending-deliveries.json"); let mut delivery = make_pending_delivery("del_legacy_deadline", "worker-a"); - let queued_at_ms = super::unix_timestamp_millis().saturating_sub(30_000); + // Derived from the effective budget, not a hard-coded 30s: the budget is + // tunable via AGENT_RELAY_DELIVERY_MAX_AGE_MS and floored at the 5s + // verification window, so a fixed offset larger than a configured budget + // would fail this assertion even though the restore logic is correct. + let budget_ms = super::delivery_budget(&MessageInjectionMode::Wait, delivery_retry_interval()) + .as_millis() as u64; + let queued_at_ms = super::unix_timestamp_millis().saturating_sub(budget_ms / 2); delivery.queued_at_ms = queued_at_ms; let deliveries = HashMap::from([(DeliveryId::new("del_legacy_deadline"), delivery)]); super::save_pending_deliveries(&path, &deliveries).expect("pending delivery should save"); @@ -3121,7 +3249,11 @@ fn legacy_pending_delivery_snapshot_rebuilds_its_acknowledgement_deadline() { let restored = &loaded["del_legacy_deadline"]; assert_eq!( restored.expires_at_ms, - super::delivery_expires_at_ms(queued_at_ms), + super::delivery_expires_at_ms( + queued_at_ms, + &MessageInjectionMode::Wait, + delivery_retry_interval() + ), "a pre-relay#1686 snapshot must come back with a deadline measured from when the \ message was queued — not unbounded, and not reset by the restart" ); @@ -3155,7 +3287,11 @@ async fn delivery_retry_success_clears_stale_last_error() { failed_attempts: 1, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: Some("old transient failure".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4084,7 +4220,11 @@ fn drop_pending_for_worker_removes_only_matching_entries() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4110,7 +4250,11 @@ fn drop_pending_for_worker_removes_only_matching_entries() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4143,7 +4287,11 @@ async fn dropped_pending_deliveries_emit_terminal_message_failures() { failed_attempts: 1, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: Some("previous blip".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4216,7 +4364,11 @@ fn should_clear_pending_delivery_when_event_id_matches() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4254,7 +4406,11 @@ fn clear_pending_delivery_returns_none_for_stale_event_id() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4674,7 +4830,11 @@ fn should_clear_pending_delivery_without_event_id_for_compatibility() { failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), - expires_at_ms: super::delivery_expires_at_ms(super::unix_timestamp_millis()), + expires_at_ms: super::delivery_expires_at_ms( + super::unix_timestamp_millis(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, diff --git a/crates/broker/src/runtime/util.rs b/crates/broker/src/runtime/util.rs index 2cc28d5556..55f31009f4 100644 --- a/crates/broker/src/runtime/util.rs +++ b/crates/broker/src/runtime/util.rs @@ -240,20 +240,27 @@ pub(crate) fn delivery_retry_interval() -> Duration { /// Wall-clock budget a single delivery gets before it is terminally failed and /// dead-lettered, defaulting to [`MAX_DELIVERY_AGE`]. /// +/// Clamped at both ends, and deliberately so. +/// /// The floor is the steer-mode verification window: a budget shorter than the /// window the broker itself waits for an echo confirmation would dead-letter -/// deliveries that are still normally in flight. There is deliberately no -/// value that disables the deadline — an unbounded delivery is the defect this -/// budget exists to close (relay#1686). +/// deliveries that are still normally in flight. (Each delivery is floored +/// again at its own acknowledgement timeout — see `delivery_budget` — so a +/// `Wait` delivery is never cut short of its 5 minute window either.) +/// +/// The ceiling exists because there must be no value that disables the +/// deadline: `u64::MAX` would otherwise be accepted as an effectively +/// unbounded age and reinstate by configuration exactly the "retries forever, +/// reports nothing" state this budget closes (relay#1686). pub(crate) fn delivery_max_age() -> Duration { let configured = std::env::var("AGENT_RELAY_DELIVERY_MAX_AGE_MS") .ok() .and_then(|raw| raw.trim().parse::().ok()) .map(Duration::from_millis) .unwrap_or(MAX_DELIVERY_AGE); - std::cmp::max( - configured, + configured.clamp( crate::broker::delivery_verification::VERIFICATION_WINDOW, + MAX_CONFIGURABLE_DELIVERY_AGE, ) } diff --git a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs index 0128a80c77..825983478a 100644 --- a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs +++ b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs @@ -80,6 +80,14 @@ const BASE_WINDOW_MS = 60_000; * never completes. */ const BODY_BYTES = 96 * 1024; +/** + * Every request is bounded, as sibling cases 1615 and 1673 do. A wedged broker + * or engine that hangs rather than refusing a connection must fail this case + * cleanly, not stall it until the 900s infrastructure timeout — a case that + * dies to the harness timeout reports nothing at all. + */ +const REQUEST_TIMEOUT_MS = 15_000; +const READINESS_TIMEOUT_MS = 2_000; const targetDir = requiredValue('RELAY_PR_PROOF_TARGET_DIR'); const harnessDir = requiredValue('RELAY_PR_PROOF_HARNESS_DIR'); @@ -133,7 +141,7 @@ try { const eng = engineClient(engineUrl); await waitFor(async () => { if (engine.exitCode !== null) throw new Error(`engine exited with code ${engine.exitCode}`); - await fetch(engineUrl); + await fetch(engineUrl, { signal: AbortSignal.timeout(READINESS_TIMEOUT_MS) }); return true; }, 'the Relaycast engine to accept connections'); @@ -211,7 +219,10 @@ try { return connection.url; }, 'the broker connection file to publish its bound API port'); const api = brokerClient(brokerUrl); - await waitFor(() => api('GET', '/api/status').then(() => true), 'the broker API to answer'); + await waitFor( + () => api('GET', '/api/status', undefined, READINESS_TIMEOUT_MS).then(() => true), + 'the broker API to answer' + ); await api('POST', '/api/spawn', { name: AGENT, @@ -224,23 +235,16 @@ try { return (listed.body?.data ?? []).some((agent) => agent.name === AGENT); }, 'the agent to register with the real engine'); - // A real DM through the engine, large enough that it cannot be written to a - // child that never reads. - const sender = await eng('POST', '/v1/agents', { name: 'proof-sender', type: 'agent' }, wsAuth); - const senderToken = sender.body?.data?.token; - if (!senderToken) { - throw new Error(`sender create failed: ${JSON.stringify(sender.body).slice(0, 300)}`); - } + // A real message through the engine, large enough that it cannot be written + // to a child that never reads. + // + // `steer`, not the default `wait`: each delivery's budget is floored at its + // own acknowledgement timeout, which is 5 minutes for `wait` and the 5 second + // verification window for `steer`. A wait-mode message would therefore ignore + // the short DEADLINE_MS this case configures and take 5 minutes per arm. const body = `relay-1686 unacked probe ${'x'.repeat(BODY_BYTES)}`; - const dm = await eng( - 'POST', - '/v1/dm', - { to: AGENT, text: body }, - { authorization: `Bearer ${senderToken}` } - ); - if (dm.status >= 300) { - throw new Error(`DM failed: ${dm.status} ${JSON.stringify(dm.body).slice(0, 300)}`); - } + const sent = await api('POST', '/api/send', { to: AGENT, text: body, mode: 'steer' }); + log(`send -> ${JSON.stringify(sent).slice(0, 200)}`); // Control, on both arms: the broker accepted the message as a retryable // delivery and has attempted it. Everything below reads as "no bound" or "a @@ -358,10 +362,11 @@ function freePort() { }); } function engineClient(baseUrl) { - return async (method, route, body, headers = {}) => { + return async (method, route, body, headers = {}, timeoutMs = REQUEST_TIMEOUT_MS) => { const res = await fetch(`${baseUrl}${route}`, { method, headers: { 'content-type': 'application/json', ...headers }, + signal: AbortSignal.timeout(timeoutMs), ...(body === undefined ? {} : { body: JSON.stringify(body) }), }); const text = await res.text(); @@ -375,10 +380,11 @@ function engineClient(baseUrl) { }; } function brokerClient(baseUrl) { - return async (method, route, body) => { + return async (method, route, body, timeoutMs = REQUEST_TIMEOUT_MS) => { const res = await fetch(`${baseUrl}${route}`, { method, headers: { 'content-type': 'application/json', 'x-api-key': BROKER_API_KEY }, + signal: AbortSignal.timeout(timeoutMs), ...(body === undefined ? {} : { body: JSON.stringify(body) }), }); const text = await res.text(); @@ -394,7 +400,9 @@ function brokerClient(baseUrl) { } async function pendingEntry(api) { const status = await api('GET', '/api/status'); - const pending = Array.isArray(status.pending) ? status.pending : []; + // `pending_deliveries`, not `pending`: reading the wrong key is what made the + // first run of this case time out on its own control. + const pending = Array.isArray(status.pending_deliveries) ? status.pending_deliveries : []; return pending.find((entry) => entry.worker_name === AGENT) ?? null; } async function deadLetter(api, deliveryId) { From 0c29b8dbcf226a4fbbd2ca0873fe22b6ab7f7e05 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 15:32:05 +0200 Subject: [PATCH 5/9] fix(broker): close follow-up delivery deadline races Session-Id: 01a07be5-64f4-7613-bc2b-98a9b82af81b --- CHANGELOG.md | 2 +- crates/broker/src/runtime/delivery.rs | 7 ++- crates/broker/src/runtime/maintenance.rs | 60 +++++++++++------- crates/broker/src/runtime/tests.rs | 80 +++++++++++++++++++++++- 4 files changed, 123 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f0af8010e2..2acea02d49 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A fleet message the broker cannot deliver to its worker is no longer reported back as handled, so it stays outstanding and can be redelivered. - Fleet deliveries the broker rejects are now logged with a reason and sequence number, so a worker that stops receiving messages can be diagnosed from the broker log. - PTY workers no longer exit when Claude Code's folder-trust dialog appears. Relay selects the affirmative option by its label, so both menu orderings work. -- A message the broker hands to an agent that never acknowledges it no longer retries forever in silence. After 30 minutes it reports `message_delivery_failed` and moves to the dead-letter store; `AGENT_RELAY_DELIVERY_MAX_AGE_MS` tunes the budget. +- A message the broker hands to an agent that never acknowledges it no longer retries forever in silence. After its acknowledgement budget (30 minutes by default), it reports `message_delivery_failed` and moves to the dead-letter store. ## [11.10.3] - 2026-09-05 diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index d604679b0b..56d4a1656a 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -1190,7 +1190,12 @@ pub(crate) fn delivery_ack_timeout( MessageInjectionMode::Wait => WAIT_DELIVERY_ACK_TIMEOUT, MessageInjectionMode::Steer => crate::broker::delivery_verification::VERIFICATION_WINDOW, }; - std::cmp::max(retry_interval, minimum) + // Retry scheduling may be configured independently, but it must not turn + // the acknowledgement floor into an effectively unbounded delivery age. + // Maintenance checks the absolute deadline independently of next_retry_at, + // so capping only this derived timeout preserves the configured handoff + // cadence while keeping the delivery budget operationally bounded. + std::cmp::max(retry_interval.min(MAX_CONFIGURABLE_DELIVERY_AGE), minimum) } pub(crate) async fn emit_delivery_attempt_outcome( diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index d9425faf5a..53e3ab6336 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -4,16 +4,18 @@ use crate::terminal_control::TerminalToCloud; impl BrokerRuntime { /// Apply worker events that are already queued, without waiting for any. - /// Returns once the channel is momentarily empty, `limit` events have been - /// applied, or the channel is closed (the `select!` arm owns that - /// transition, so it is simply left alone here). - async fn drain_ready_worker_events(&mut self, limit: usize) { + /// Returns `true` only when the channel is momentarily empty or closed. If + /// the bound is reached with events still queued, the caller must defer any + /// delivery retry/deadline sweep: a later queued event may be the + /// confirmation for a delivery that otherwise looks expired. + async fn drain_ready_worker_events(&mut self, limit: usize) -> bool { for _ in 0..limit { match self.worker_event_rx.try_recv() { Ok(event) => self.handle_worker_event(event).await, - Err(_) => return, + Err(_) => return true, } } + self.worker_event_rx.is_empty() } pub(super) async fn handle_maintenance_tick(&mut self) { @@ -27,9 +29,13 @@ impl BrokerRuntime { // message the agent already read. Draining first makes the sweep read // the freshest state the broker actually has. See relay#1686. // - // Bounded so a busy worker cannot starve the rest of the tick; anything - // left is handled by the normal `select!` arm on the next iterations. - self.drain_ready_worker_events(MAX_DRAINED_WORKER_EVENTS_PER_TICK) + // Bounded so a busy worker cannot starve the rest of the tick. If the + // bound is reached while events remain, the rest of maintenance still + // runs but the delivery sweep is deferred: a confirmation behind the + // bound must win over an apparent deadline. Anything left is handled + // by the normal `select!` arm or the next tick. + let worker_events_drained = self + .drain_ready_worker_events(MAX_DRAINED_WORKER_EVENTS_PER_TICK) .await; let paths = &self.paths; @@ -184,21 +190,29 @@ impl BrokerRuntime { // hold, but late. `retry_pending_delivery` still owns the decision; // this only decides when it gets asked. See relay#1686. let now_ms = unix_timestamp_millis(); - let due_ids: Vec = pending_deliveries - .iter() - .filter_map(|(delivery_id, pending)| { - let confirmation_is_held = - pending.withheld_fleet_ack.as_ref().is_some_and(|deliver| { - fleet_delivery_book.is_delivery_confirmation_held(deliver) - }); - let past_deadline = now_ms >= pending.expires_at_ms; - if (pending.next_retry_at <= now || past_deadline) && !confirmation_is_held { - Some(delivery_id.clone()) - } else { - None - } - }) - .collect(); + let due_ids: Vec = if worker_events_drained { + pending_deliveries + .iter() + .filter_map(|(delivery_id, pending)| { + let confirmation_is_held = + pending.withheld_fleet_ack.as_ref().is_some_and(|deliver| { + fleet_delivery_book.is_delivery_confirmation_held(deliver) + }); + let past_deadline = now_ms >= pending.expires_at_ms; + if (pending.next_retry_at <= now || past_deadline) && !confirmation_is_held { + Some(delivery_id.clone()) + } else { + None + } + }) + .collect() + } else { + tracing::debug!( + target = "relay_broker::delivery", + "deferring delivery sweep until queued worker events are applied" + ); + Vec::new() + }; for delivery_id in due_ids { let was_retry = pending_deliveries diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index d894f66ec6..fd26839c4a 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -200,6 +200,7 @@ async fn cleanup_worker_registry(mut registry: WorkerRegistry) { struct WorkerEventRuntimeFixture { runtime: BrokerRuntime, + worker_event_tx: mpsc::Sender, fleet_control_rx: mpsc::Receiver, _sdk_out_rx: mpsc::Receiver>, _temp_dir: tempfile::TempDir, @@ -234,7 +235,7 @@ fn worker_event_runtime_fixture( let (terminal_control_tx, _terminal_control_rx) = mpsc::channel(4); let (_terminal_event_tx, terminal_event_rx) = mpsc::channel(4); let (sdk_out_tx, sdk_out_rx) = mpsc::channel(64); - let (_worker_event_tx, worker_event_rx) = mpsc::channel(4); + let (worker_event_tx, worker_event_rx) = mpsc::channel(1024); let (hosted_agent_event_tx, _hosted_agent_event_rx) = mpsc::channel(4); let mut reap_tick = tokio::time::interval(Duration::from_secs(60)); reap_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -313,6 +314,7 @@ fn worker_event_runtime_fixture( WorkerEventRuntimeFixture { runtime, + worker_event_tx, fleet_control_rx, _sdk_out_rx: sdk_out_rx, _temp_dir: temp_dir, @@ -3081,6 +3083,69 @@ async fn unacked_delivery_terminates_on_the_cumulative_attempt_ceiling() { ); } +// relay#1686 review follow-up: a maintenance tick may observe an expired +// delivery while its matching confirmation is queued behind a burst of other +// worker events. Reaching the bounded drain limit must defer the delivery sweep +// rather than dead-lettering before the later confirmation is applied. +#[tokio::test] +async fn partial_worker_event_drain_defers_the_delivery_sweep() { + let worker_name = "worker-busy-confirmation"; + let registry = make_worker_registry_with_worker(worker_name).await; + let generation = registry.workers[worker_name].generation; + let delivery_id = "del_queued_confirmation"; + let mut pending = make_pending_delivery(delivery_id, worker_name); + pending.expires_at_ms = 0; + let mut fixture = worker_event_runtime_fixture( + registry, + HashMap::from([(DeliveryId::new(delivery_id), pending)]), + ); + + for index in 0..super::MAX_DRAINED_WORKER_EVENTS_PER_TICK { + fixture + .worker_event_tx + .try_send(delivery_lifecycle_worker_event( + worker_name, + generation, + "test_queue_noise", + &format!("noise-{index}"), + &format!("evt-noise-{index}"), + )) + .expect("worker event burst should fit the production-sized test channel"); + } + fixture + .worker_event_tx + .try_send(delivery_lifecycle_worker_event( + worker_name, + generation, + "delivery_verified", + delivery_id, + &format!("evt_{delivery_id}"), + )) + .expect("the matching confirmation should be queued behind the drain bound"); + + fixture.runtime.handle_maintenance_tick().await; + assert!( + fixture.runtime.pending_deliveries.contains_key(delivery_id), + "an expired delivery must remain pending while its confirmation may still be queued" + ); + assert!( + fixture.runtime.dead_letters.is_empty(), + "a partial event drain must not dead-letter before queued confirmations are applied" + ); + + fixture.runtime.handle_maintenance_tick().await; + assert!( + !fixture.runtime.pending_deliveries.contains_key(delivery_id), + "the next tick must apply the queued confirmation" + ); + assert!( + fixture.runtime.dead_letters.is_empty(), + "a confirmed delivery must never be dead-lettered" + ); + + cleanup_worker_registry(fixture.runtime.workers).await; +} + // relay#1686 review follow-up: the attempt ceiling is a clock-skew backstop, so // it must never be what ends a delivery while the wall-clock deadline is still // in the future. A fixed ceiling could not hold that: at 1000 attempts a `Steer` @@ -3147,6 +3212,18 @@ fn configured_delivery_age_is_clamped_at_both_ends() { super::delivery_budget(&MessageInjectionMode::Steer, delivery_retry_interval()), crate::broker::delivery_verification::VERIFICATION_WINDOW, ); + + let oversized_retry_interval = MAX_CONFIGURABLE_DELIVERY_AGE + Duration::from_secs(1); + assert_eq!( + super::delivery_ack_timeout(&MessageInjectionMode::Steer, oversized_retry_interval), + MAX_CONFIGURABLE_DELIVERY_AGE, + "an oversized retry interval must not extend the acknowledgement timeout past the cap" + ); + assert_eq!( + super::delivery_budget(&MessageInjectionMode::Steer, oversized_retry_interval), + MAX_CONFIGURABLE_DELIVERY_AGE, + "an oversized retry interval must not extend the delivery budget past the cap" + ); std::env::remove_var("AGENT_RELAY_DELIVERY_MAX_AGE_MS"); } @@ -3218,6 +3295,7 @@ fn requeued_dead_letter_gets_a_fresh_acknowledgement_budget() { // hours does not get its clock reset by the restart either. #[test] fn legacy_pending_delivery_snapshot_rebuilds_its_acknowledgement_deadline() { + let _guard = env_test_lock().lock().expect("env test lock"); let dir = tempfile::tempdir().expect("tempdir should create"); let path = dir.path().join("pending-deliveries.json"); let mut delivery = make_pending_delivery("del_legacy_deadline", "worker-a"); From c806a73edf499306b8df71c875763a55150220d9 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 16:08:36 +0200 Subject: [PATCH 6/9] test(pr-proof): fit delivery proof inside Cloud step Session-Id: 01a07be5-64f4-7613-bc2b-98a9b82af81b --- .../run.mjs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs index 825983478a..d57e80f5a6 100644 --- a/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs +++ b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs @@ -64,16 +64,18 @@ const READY_TIMEOUT_MS = 90_000; const DEADLINE_MS = 10_000; /** * How long the head broker is given to act on that budget. The deadline is - * swept on the broker's 500ms maintenance tick, so this is generous by an order - * of magnitude. + * swept on the broker's 500ms maintenance tick, so three times the configured + * budget leaves ample scheduling margin. */ -const HEAD_WINDOW_MS = 45_000; +const HEAD_WINDOW_MS = 30_000; /** - * How long the base broker is watched for any terminal outcome. Six times the - * deadline and four times the head window: if a bound existed anywhere in the - * base broker's retry path, it would have fired well inside this. + * How long the base broker is watched for any terminal outcome. Twice the + * configured deadline is enough to prove that the base broker ignores that + * bound while still observing several 5-second steer retry cycles. Keeping the + * window focused also leaves the Cloud proof owner enough time for its clean + * checkout, engine install, and broker startup before the step deadline. */ -const BASE_WINDOW_MS = 60_000; +const BASE_WINDOW_MS = 20_000; /** * Message body size. The tty input queue is a few kilobytes, so a body this * size cannot be written to a child that never reads, and the injection write From 9db26fc7a71bf9b9bbdb57436c94628f9be9bcaf Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 16:23:56 +0200 Subject: [PATCH 7/9] fix(broker): order delivery sweep behind queued events Session-Id: 01a07be5-64f4-7613-bc2b-98a9b82af81b --- crates/broker/src/runtime/maintenance.rs | 70 +++++++++++------------- crates/broker/src/runtime/mod.rs | 3 - crates/broker/src/runtime/tests.rs | 55 +++++++++++++++---- 3 files changed, 75 insertions(+), 53 deletions(-) diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 53e3ab6336..18d1b65fe2 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -3,19 +3,21 @@ use super::*; use crate::terminal_control::TerminalToCloud; impl BrokerRuntime { - /// Apply worker events that are already queued, without waiting for any. - /// Returns `true` only when the channel is momentarily empty or closed. If - /// the bound is reached with events still queued, the caller must defer any - /// delivery retry/deadline sweep: a later queued event may be the - /// confirmation for a delivery that otherwise looks expired. - async fn drain_ready_worker_events(&mut self, limit: usize) -> bool { - for _ in 0..limit { + /// Apply the FIFO prefix of worker events that was queued when maintenance + /// began, without waiting for anything produced afterward. + /// + /// The receiver is the only consumer, so its starting length is an ordering + /// barrier: every confirmation preceding the tick is inside this finite + /// prefix. Events appended concurrently remain for the normal `select!` + /// arm, while delivery maintenance can proceed without waiting for a busy + /// channel to become empty. + async fn drain_queued_worker_event_prefix(&mut self, queued_before_tick: usize) { + for _ in 0..queued_before_tick { match self.worker_event_rx.try_recv() { Ok(event) => self.handle_worker_event(event).await, - Err(_) => return true, + Err(_) => return, } } - self.worker_event_rx.is_empty() } pub(super) async fn handle_maintenance_tick(&mut self) { @@ -29,13 +31,11 @@ impl BrokerRuntime { // message the agent already read. Draining first makes the sweep read // the freshest state the broker actually has. See relay#1686. // - // Bounded so a busy worker cannot starve the rest of the tick. If the - // bound is reached while events remain, the rest of maintenance still - // runs but the delivery sweep is deferred: a confirmation behind the - // bound must win over an apparent deadline. Anything left is handled - // by the normal `select!` arm or the next tick. - let worker_events_drained = self - .drain_ready_worker_events(MAX_DRAINED_WORKER_EVENTS_PER_TICK) + // Snapshot the FIFO prefix instead of draining until empty. This is an + // ordering barrier for every confirmation already queued, but traffic + // arriving during the drain cannot starve delivery maintenance. + let queued_worker_events = self.worker_event_rx.len(); + self.drain_queued_worker_event_prefix(queued_worker_events) .await; let paths = &self.paths; @@ -190,29 +190,21 @@ impl BrokerRuntime { // hold, but late. `retry_pending_delivery` still owns the decision; // this only decides when it gets asked. See relay#1686. let now_ms = unix_timestamp_millis(); - let due_ids: Vec = if worker_events_drained { - pending_deliveries - .iter() - .filter_map(|(delivery_id, pending)| { - let confirmation_is_held = - pending.withheld_fleet_ack.as_ref().is_some_and(|deliver| { - fleet_delivery_book.is_delivery_confirmation_held(deliver) - }); - let past_deadline = now_ms >= pending.expires_at_ms; - if (pending.next_retry_at <= now || past_deadline) && !confirmation_is_held { - Some(delivery_id.clone()) - } else { - None - } - }) - .collect() - } else { - tracing::debug!( - target = "relay_broker::delivery", - "deferring delivery sweep until queued worker events are applied" - ); - Vec::new() - }; + let due_ids: Vec = pending_deliveries + .iter() + .filter_map(|(delivery_id, pending)| { + let confirmation_is_held = + pending.withheld_fleet_ack.as_ref().is_some_and(|deliver| { + fleet_delivery_book.is_delivery_confirmation_held(deliver) + }); + let past_deadline = now_ms >= pending.expires_at_ms; + if (pending.next_retry_at <= now || past_deadline) && !confirmation_is_held { + Some(delivery_id.clone()) + } else { + None + } + }) + .collect(); for delivery_id in due_ids { let was_retry = pending_deliveries diff --git a/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs index 3f95ad2dc7..41abe87ee0 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -112,9 +112,6 @@ const ATTEMPT_CEILING_HEADROOM: u32 = 4; /// wall-clock termination guarantee this whole change exists to provide — the /// same "retries forever, reports nothing" state, reachable by configuration. const MAX_CONFIGURABLE_DELIVERY_AGE: Duration = Duration::from_secs(7 * 24 * 60 * 60); -/// How many already-queued worker events a maintenance tick applies before the -/// rest of its work. Bounded so a chatty worker cannot starve the tick. -const MAX_DRAINED_WORKER_EVENTS_PER_TICK: usize = 256; const THREAD_HISTORY_LIMIT: usize = 1_000; #[allow(dead_code)] // only http_api_local_delivery_timeout's default; see its own allow const DEFAULT_HTTP_API_LOCAL_DELIVERY_TIMEOUT_MS: u64 = 3_000; diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index fd26839c4a..b5cbf46d3c 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -3085,10 +3085,10 @@ async fn unacked_delivery_terminates_on_the_cumulative_attempt_ceiling() { // relay#1686 review follow-up: a maintenance tick may observe an expired // delivery while its matching confirmation is queued behind a burst of other -// worker events. Reaching the bounded drain limit must defer the delivery sweep -// rather than dead-lettering before the later confirmation is applied. +// worker events. The tick-start FIFO prefix is an ordering barrier: even a +// confirmation far behind the old 256-event limit must apply before the sweep. #[tokio::test] -async fn partial_worker_event_drain_defers_the_delivery_sweep() { +async fn queued_worker_event_prefix_applies_confirmation_before_delivery_sweep() { let worker_name = "worker-busy-confirmation"; let registry = make_worker_registry_with_worker(worker_name).await; let generation = registry.workers[worker_name].generation; @@ -3100,7 +3100,7 @@ async fn partial_worker_event_drain_defers_the_delivery_sweep() { HashMap::from([(DeliveryId::new(delivery_id), pending)]), ); - for index in 0..super::MAX_DRAINED_WORKER_EVENTS_PER_TICK { + for index in 0..512 { fixture .worker_event_tx .try_send(delivery_lifecycle_worker_event( @@ -3121,26 +3121,59 @@ async fn partial_worker_event_drain_defers_the_delivery_sweep() { delivery_id, &format!("evt_{delivery_id}"), )) - .expect("the matching confirmation should be queued behind the drain bound"); + .expect("the matching confirmation should be queued deep in the FIFO prefix"); fixture.runtime.handle_maintenance_tick().await; assert!( - fixture.runtime.pending_deliveries.contains_key(delivery_id), - "an expired delivery must remain pending while its confirmation may still be queued" + !fixture.runtime.pending_deliveries.contains_key(delivery_id), + "the tick must apply the queued confirmation before examining the expired delivery" ); assert!( fixture.runtime.dead_letters.is_empty(), - "a partial event drain must not dead-letter before queued confirmations are applied" + "a confirmed delivery must never be dead-lettered" + ); + + cleanup_worker_registry(fixture.runtime.workers).await; +} + +// The ordering barrier is a finite snapshot, not a queue-emptiness gate. A +// continuously busy worker therefore cannot postpone retries or terminal +// deadline handling: after the prefix that preceded this tick is applied, the +// delivery sweep always runs even if producers have more work to append. +#[tokio::test] +async fn full_worker_event_backlog_does_not_suppress_delivery_expiry() { + let worker_name = "worker-sustained-backlog"; + let registry = make_worker_registry_with_worker(worker_name).await; + let generation = registry.workers[worker_name].generation; + let delivery_id = "del_expired_under_backlog"; + let mut pending = make_pending_delivery(delivery_id, worker_name); + pending.expires_at_ms = 0; + let mut fixture = worker_event_runtime_fixture( + registry, + HashMap::from([(DeliveryId::new(delivery_id), pending)]), ); + for index in 0..1024 { + fixture + .worker_event_tx + .try_send(delivery_lifecycle_worker_event( + worker_name, + generation, + "test_queue_noise", + &format!("noise-{index}"), + &format!("evt-noise-{index}"), + )) + .expect("the worker event backlog should fill the test channel"); + } + fixture.runtime.handle_maintenance_tick().await; assert!( !fixture.runtime.pending_deliveries.contains_key(delivery_id), - "the next tick must apply the queued confirmation" + "a full worker-event backlog must not suppress terminal delivery maintenance" ); assert!( - fixture.runtime.dead_letters.is_empty(), - "a confirmed delivery must never be dead-lettered" + fixture.runtime.dead_letters.get(delivery_id).is_some(), + "the expired delivery must reach the dead-letter store under sustained traffic" ); cleanup_worker_registry(fixture.runtime.workers).await; From 7cb1b52e763f0a9eede9ca20ae443435f46b21e0 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 16:42:02 +0200 Subject: [PATCH 8/9] fix(broker): linearize delivery expiry behind worker events Session-Id: 01a07be5-64f4-7613-bc2b-98a9b82af81b --- crates/broker/src/runtime/maintenance.rs | 55 +++++++++++++++------- crates/broker/src/runtime/tests.rs | 13 ++--- crates/broker/src/runtime/worker_events.rs | 5 ++ crates/broker/src/worker.rs | 12 +++++ 4 files changed, 61 insertions(+), 24 deletions(-) diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 18d1b65fe2..ab00b087d3 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -3,19 +3,39 @@ use super::*; use crate::terminal_control::TerminalToCloud; impl BrokerRuntime { - /// Apply the FIFO prefix of worker events that was queued when maintenance - /// began, without waiting for anything produced afterward. + /// Establish a FIFO linearization point, then apply every worker event that + /// acquired channel capacity before it. /// - /// The receiver is the only consumer, so its starting length is an ordering - /// barrier: every confirmation preceding the tick is inside this finite - /// prefix. Events appended concurrently remain for the normal `select!` - /// arm, while delivery maintenance can proceed without waiting for a busy - /// channel to become empty. - async fn drain_queued_worker_event_prefix(&mut self, queued_before_tick: usize) { - for _ in 0..queued_before_tick { - match self.worker_event_rx.try_recv() { - Ok(event) => self.handle_worker_event(event).await, - Err(_) => return, + /// Reserving a slot is important when the bounded channel is full: the + /// reservation joins the send queue while this method continues receiving, + /// so sustained producers cannot take every newly freed slot and starve the + /// marker. Once inserted, normal FIFO order means confirmations before the + /// marker are applied before delivery expiry and events after it belong to + /// the next actor turn. + async fn drain_worker_events_through_maintenance_barrier(&mut self) { + let reserve = self.workers.event_sender().reserve_owned(); + tokio::pin!(reserve); + + let permit = loop { + tokio::select! { + biased; + result = &mut reserve => match result { + Ok(permit) => break permit, + Err(_) => return, + }, + event = self.worker_event_rx.recv() => match event { + Some(WorkerEvent::MaintenanceBarrier) => continue, + Some(event) => self.handle_worker_event(event).await, + None => return, + }, + } + }; + permit.send(WorkerEvent::MaintenanceBarrier); + + while let Some(event) = self.worker_event_rx.recv().await { + match event { + WorkerEvent::MaintenanceBarrier => return, + event => self.handle_worker_event(event).await, } } } @@ -31,12 +51,11 @@ impl BrokerRuntime { // message the agent already read. Draining first makes the sweep read // the freshest state the broker actually has. See relay#1686. // - // Snapshot the FIFO prefix instead of draining until empty. This is an - // ordering barrier for every confirmation already queued, but traffic - // arriving during the drain cannot starve delivery maintenance. - let queued_worker_events = self.worker_event_rx.len(); - self.drain_queued_worker_event_prefix(queued_worker_events) - .await; + // Insert a marker through the same FIFO as worker confirmations and + // drain through it. The marker supplies an exact ordering boundary even + // when a confirmation arrives while an earlier backlog is being + // handled; traffic ordered after it cannot starve the sweep. + self.drain_worker_events_through_maintenance_barrier().await; let paths = &self.paths; let state = &mut self.state; diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index b5cbf46d3c..488734b308 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -207,7 +207,7 @@ struct WorkerEventRuntimeFixture { } fn worker_event_runtime_fixture( - workers: WorkerRegistry, + mut workers: WorkerRegistry, pending_deliveries: HashMap, ) -> WorkerEventRuntimeFixture { let temp_dir = tempfile::tempdir().expect("runtime fixture temp dir"); @@ -236,6 +236,7 @@ fn worker_event_runtime_fixture( let (_terminal_event_tx, terminal_event_rx) = mpsc::channel(4); let (sdk_out_tx, sdk_out_rx) = mpsc::channel(64); let (worker_event_tx, worker_event_rx) = mpsc::channel(1024); + workers.set_event_sender_for_test(worker_event_tx.clone()); let (hosted_agent_event_tx, _hosted_agent_event_rx) = mpsc::channel(4); let mut reap_tick = tokio::time::interval(Duration::from_secs(60)); reap_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -3085,7 +3086,7 @@ async fn unacked_delivery_terminates_on_the_cumulative_attempt_ceiling() { // relay#1686 review follow-up: a maintenance tick may observe an expired // delivery while its matching confirmation is queued behind a burst of other -// worker events. The tick-start FIFO prefix is an ordering barrier: even a +// worker events. A marker in the same FIFO is the ordering barrier: even a // confirmation far behind the old 256-event limit must apply before the sweep. #[tokio::test] async fn queued_worker_event_prefix_applies_confirmation_before_delivery_sweep() { @@ -3136,10 +3137,10 @@ async fn queued_worker_event_prefix_applies_confirmation_before_delivery_sweep() cleanup_worker_registry(fixture.runtime.workers).await; } -// The ordering barrier is a finite snapshot, not a queue-emptiness gate. A -// continuously busy worker therefore cannot postpone retries or terminal -// deadline handling: after the prefix that preceded this tick is applied, the -// delivery sweep always runs even if producers have more work to append. +// The ordering barrier reserves a finite position, rather than waiting for +// queue emptiness. A continuously busy worker therefore cannot postpone +// retries or terminal deadline handling: after the marker is reached, the +// delivery sweep runs even if producers have more work to append. #[tokio::test] async fn full_worker_event_backlog_does_not_suppress_delivery_expiry() { let worker_name = "worker-sustained-backlog"; diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 0132b590b7..28311c2df3 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -629,6 +629,11 @@ impl BrokerRuntime { let terminal_input_requests = &mut self.terminal_input_requests; match worker_event { + WorkerEvent::MaintenanceBarrier => { + // Maintenance consumes its own barrier before dispatch. Keep a + // defensive no-op so shutdown/cancellation cannot make an + // internal ordering marker observable as a worker failure. + } WorkerEvent::WriterFailed { name, generation, diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index d99ff9172c..f985af0cd9 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -223,6 +223,9 @@ impl AgentWorkState { #[derive(Debug, Clone)] pub(crate) enum WorkerEvent { + /// Runtime-internal FIFO barrier used to order maintenance after every + /// worker event that acquired channel capacity before the barrier. + MaintenanceBarrier, Message { name: WorkerName, generation: Uuid, @@ -364,6 +367,15 @@ impl WorkerRegistry { } } + pub(crate) fn event_sender(&self) -> mpsc::Sender { + self.event_tx.clone() + } + + #[cfg(test)] + pub(crate) fn set_event_sender_for_test(&mut self, event_tx: mpsc::Sender) { + self.event_tx = event_tx; + } + fn commit_hooks_dir(&mut self) -> Result<&Path> { resolve_commit_hooks_dir(&mut self.commit_hooks_dir) } From 27592c5418f61425e34debfd017e950fa3405c57 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Mon, 7 Sep 2026 16:57:04 +0200 Subject: [PATCH 9/9] chore(trajectories): record the relay#1686 delivery-deadline trajectory Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014CiShnEAJ5Trb7JovG8SdG Session-Id: 2ceeba77-2877-4b1e-b973-bf0453a2e37b --- .../2026-09/traj_utmpf8clg0x2.trace.json | 378 ++++++++++++++++++ .../2026-09/traj_utmpf8clg0x2/summary.md | 44 ++ .../2026-09/traj_utmpf8clg0x2/trajectory.json | 87 ++++ 3 files changed, 509 insertions(+) create mode 100644 .agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2.trace.json create mode 100644 .agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/summary.md create mode 100644 .agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/trajectory.json diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2.trace.json b/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2.trace.json new file mode 100644 index 0000000000..bf4b9e1b2d --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2.trace.json @@ -0,0 +1,378 @@ +{ + "version": "1.0.0", + "id": "9faa2399-ce4a-4012-ace0-b95b4c2f65c0", + "timestamp": "2026-09-07T14:56:28.208Z", + "trajectory": "traj_utmpf8clg0x2", + "files": [ + { + "path": "CHANGELOG.md", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 9, + "end_line": 16, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 22, + "end_line": 28, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/runtime/dead_letter.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 219, + "end_line": 234, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 236, + "end_line": 242, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/runtime/delivery.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 55, + "end_line": 67, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 168, + "end_line": 237, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 323, + "end_line": 329, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 350, + "end_line": 356, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 360, + "end_line": 379, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 999, + "end_line": 1006, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 1010, + "end_line": 1016, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 1043, + "end_line": 1070, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 1117, + "end_line": 1123, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 1190, + "end_line": 1201, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/runtime/maintenance.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 3, + "end_line": 62, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/runtime/mod.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 86, + "end_line": 117, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/runtime/tests.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 56, + "end_line": 64, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 200, + "end_line": 213, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 235, + "end_line": 242, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 315, + "end_line": 321, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 412, + "end_line": 422, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 1135, + "end_line": 1145, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 2562, + "end_line": 2572, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 2688, + "end_line": 2698, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 2835, + "end_line": 2849, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 2864, + "end_line": 2874, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 2879, + "end_line": 2885, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 3022, + "end_line": 3028, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 3084, + "end_line": 3287, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 3294, + "end_line": 3304, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 3329, + "end_line": 3345, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 3361, + "end_line": 3371, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 3399, + "end_line": 3409, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 4332, + "end_line": 4342, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 4362, + "end_line": 4372, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 4399, + "end_line": 4409, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 4476, + "end_line": 4486, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 4518, + "end_line": 4528, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 4942, + "end_line": 4952, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/runtime/util.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 240, + "end_line": 266, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/runtime/worker_events.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 629, + "end_line": 639, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "crates/broker/src/worker.rs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 223, + "end_line": 231, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 367, + "end_line": 381, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + }, + { + "path": "tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 64, + "end_line": 95, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 143, + "end_line": 149, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 221, + "end_line": 230, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 237, + "end_line": 252, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 364, + "end_line": 374, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 382, + "end_line": 392, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + }, + { + "start_line": 402, + "end_line": 410, + "revision": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0" + } + ] + } + ] + } + ] +} \ No newline at end of file diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/summary.md b/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/summary.md new file mode 100644 index 0000000000..8157ee132b --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/summary.md @@ -0,0 +1,44 @@ +# Trajectory: relay#1686: bound an unACKed delivery and give it a terminal dead-letter path + +> **Status:** ✅ Completed +> **Task:** relay#1686 +> **Confidence:** 85% +> **Started:** September 7, 2026 at 12:52 PM +> **Completed:** September 7, 2026 at 04:56 PM + +--- + +## Summary + +relay#1686: bounded an unACKed delivery with a per-delivery wall-clock acknowledgement deadline (30 min default, AGENT_RELAY_DELIVERY_MAX_AGE_MS, clamped both ends, floored per delivery at its own ack timeout) plus a budget-scaled cumulative attempt ceiling as a clock-skew backstop. Both drive the existing terminal path: message_delivery_failed plus a dead-letter entry, withheld fleet ack dropped. failed_attempts cap untouched. Mutation-proven twice; both arms verified end-to-end against real broker binaries and a real Relaycast engine. 16 of 18 review findings fixed, 2 skipped with reason. PR #1701, not merged. + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts +- **Chose:** Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts +- **Reasoning:** The retry cadence IS delivery_ack_timeout: 5 min in Wait mode, the 5s steer verification window. The same attempt count therefore means ~50 minutes in one mode and ~50 seconds in the other, so no single attempt number both spares an agent mid-turn and catches a deaf recipient. 30 min is 6x the wait-mode ack timeout and 360x the steer window. A cumulative attempts ceiling rides along only as a clock-skew backstop, scaled from the configured budget so it can never preempt a raised deadline. + +### A healthy PTY worker cannot reproduce a never-ACKed delivery +- **Chose:** A healthy PTY worker cannot reproduce a never-ACKed delivery +- **Reasoning:** pty_worker acks on echo verification OR a 5s timeout fallback (MAX_VERIFICATION_ATTEMPTS=1), so no badly-behaved child can suppress the ack, and the tty line discipline echoes input regardless of stty. The ack is withheld only while the injection WRITE has not completed. The proof case therefore wedges the write itself: a child that never reads stdin plus a body larger than the tty input queue. Relevant to the whole deaf-agent cluster (#1670, #1689): look at the write path, not the child. + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts: Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts +- A healthy PTY worker cannot reproduce a never-ACKed delivery: A healthy PTY worker cannot reproduce a never-ACKed delivery + +--- + +## Artifacts + +**Commits:** 7cb1b52e7, 9db26fc7a, c806a73ed, 0c29b8dbc, 4466f9aff +**Files changed:** 10 diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/trajectory.json b/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/trajectory.json new file mode 100644 index 0000000000..193b42fd7d --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_utmpf8clg0x2/trajectory.json @@ -0,0 +1,87 @@ +{ + "id": "traj_utmpf8clg0x2", + "version": 1, + "task": { + "title": "relay#1686: bound an unACKed delivery and give it a terminal dead-letter path", + "source": { + "system": "plain", + "id": "relay#1686" + } + }, + "status": "completed", + "startedAt": "2026-09-07T10:52:41.583Z", + "completedAt": "2026-09-07T14:56:27.871Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-09-07T14:55:41.854Z" + } + ], + "chapters": [ + { + "id": "chap_sicop27lpxk0", + "title": "Work", + "agentName": "default", + "startedAt": "2026-09-07T14:55:41.854Z", + "endedAt": "2026-09-07T14:56:27.871Z", + "events": [ + { + "ts": 1788792941855, + "type": "decision", + "content": "Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts: Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts", + "raw": { + "question": "Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts", + "chosen": "Bound the unACKed delivery with a wall-clock deadline, not a cap on attempts", + "alternatives": [], + "reasoning": "The retry cadence IS delivery_ack_timeout: 5 min in Wait mode, the 5s steer verification window. The same attempt count therefore means ~50 minutes in one mode and ~50 seconds in the other, so no single attempt number both spares an agent mid-turn and catches a deaf recipient. 30 min is 6x the wait-mode ack timeout and 360x the steer window. A cumulative attempts ceiling rides along only as a clock-skew backstop, scaled from the configured budget so it can never preempt a raised deadline." + }, + "significance": "high" + }, + { + "ts": 1788792962826, + "type": "decision", + "content": "A healthy PTY worker cannot reproduce a never-ACKed delivery: A healthy PTY worker cannot reproduce a never-ACKed delivery", + "raw": { + "question": "A healthy PTY worker cannot reproduce a never-ACKed delivery", + "chosen": "A healthy PTY worker cannot reproduce a never-ACKed delivery", + "alternatives": [], + "reasoning": "pty_worker acks on echo verification OR a 5s timeout fallback (MAX_VERIFICATION_ATTEMPTS=1), so no badly-behaved child can suppress the ack, and the tty line discipline echoes input regardless of stty. The ack is withheld only while the injection WRITE has not completed. The proof case therefore wedges the write itself: a child that never reads stdin plus a body larger than the tty input queue. Relevant to the whole deaf-agent cluster (#1670, #1689): look at the write path, not the child." + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "relay#1686: bounded an unACKed delivery with a per-delivery wall-clock acknowledgement deadline (30 min default, AGENT_RELAY_DELIVERY_MAX_AGE_MS, clamped both ends, floored per delivery at its own ack timeout) plus a budget-scaled cumulative attempt ceiling as a clock-skew backstop. Both drive the existing terminal path: message_delivery_failed plus a dead-letter entry, withheld fleet ack dropped. failed_attempts cap untouched. Mutation-proven twice; both arms verified end-to-end against real broker binaries and a real Relaycast engine. 16 of 18 review findings fixed, 2 skipped with reason. PR #1701, not merged.", + "approach": "Standard approach", + "confidence": 0.85 + }, + "commits": [ + "7cb1b52e7", + "9db26fc7a", + "c806a73ed", + "0c29b8dbc", + "4466f9aff" + ], + "filesChanged": [ + "CHANGELOG.md", + "crates/broker/src/runtime/dead_letter.rs", + "crates/broker/src/runtime/delivery.rs", + "crates/broker/src/runtime/maintenance.rs", + "crates/broker/src/runtime/mod.rs", + "crates/broker/src/runtime/tests.rs", + "crates/broker/src/runtime/util.rs", + "crates/broker/src/runtime/worker_events.rs", + "crates/broker/src/worker.rs", + "tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs" + ], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "b0db16b7cee8f7d776ea783a3002f2a59393a3d5", + "endRef": "7cb1b52e763f0a9eede9ca20ae443435f46b21e0", + "traceId": "9faa2399-ce4a-4012-ace0-b95b4c2f65c0" + } +} \ No newline at end of file