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 diff --git a/CHANGELOG.md b/CHANGELOG.md index cd719125de..2acea02d49 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased - Minor] +### Added + +- `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 - `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 +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 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/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..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,6 +236,7 @@ pub(crate) fn requeue_dead_letter( failed_attempts: 0, next_retry_at: Instant::now(), queued_at_ms: entry.queued_at_ms, + 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 abbe5f9582..56d4a1656a 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,13 @@ pub(crate) struct PersistedPendingDelivery { pub(super) failed_attempts: u32, #[serde(default)] pub(super) queued_at_ms: u64, + /// 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: Option, #[serde(default)] pub(super) last_error: Option, /// See `PendingDelivery::withheld_fleet_ack`. `#[serde(default)]` so a @@ -148,6 +168,72 @@ 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 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, /// remove, retry bookkeeping) marks the store dirty via `DerefMut`, letting /// the event loop persist the snapshot immediately after the mutating event @@ -237,6 +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: 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, @@ -258,6 +345,12 @@ 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; + 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 {budget_secs}s acknowledgement deadline", + pending.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 {ceiling}", + pending.attempts + )); + } + None +} + pub(crate) async fn retry_pending_delivery( delivery_id: &DeliveryId, workers: &mut WorkerRegistry, @@ -955,6 +1102,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, retry_interval) { + 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 @@ -1011,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 2d69f15476..ab00b087d3 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -3,7 +3,60 @@ use super::*; use crate::terminal_control::TerminalToCloud; impl BrokerRuntime { + /// Establish a FIFO linearization point, then apply every worker event that + /// acquired channel capacity before it. + /// + /// 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, + } + } + } + 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. + // + // 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; let sdk_out_tx = &self.sdk_out_tx; @@ -149,6 +202,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 +216,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/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs index 90068ffa87..41abe87ee0 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -58,8 +58,60 @@ 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); +/// Floor for the *cumulative* attempt ceiling, which is otherwise scaled from +/// the configured budget (see `delivery_attempt_ceiling`). +/// +/// 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); 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..488734b308 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_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::{ @@ -199,13 +200,14 @@ 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, } fn worker_event_runtime_fixture( - workers: WorkerRegistry, + mut workers: WorkerRegistry, pending_deliveries: HashMap, ) -> WorkerEventRuntimeFixture { let temp_dir = tempfile::tempdir().expect("runtime fixture temp dir"); @@ -233,7 +235,8 @@ 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); + 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); @@ -312,6 +315,7 @@ fn worker_event_runtime_fixture( WorkerEventRuntimeFixture { runtime, + worker_event_tx, fleet_control_rx, _sdk_out_rx: sdk_out_rx, _temp_dir: temp_dir, @@ -408,6 +412,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -1126,6 +1135,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2548,6 +2562,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: Some("failed writing frame".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2669,6 +2688,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -2787,6 +2811,570 @@ 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. + // 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 { + 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, + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), + 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..600 { + 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: u32::MAX, + // 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"); + + 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 { + unreachable!("asserted Failed above"); + }; + 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 review follow-up: a maintenance tick may observe an expired +// delivery while its matching confirmation is queued behind a burst of other +// 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() { + 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..512 { + 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 deep in the FIFO prefix"); + + fixture.runtime.handle_maintenance_tick().await; + assert!( + !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 confirmed delivery must never be dead-lettered" + ); + + cleanup_worker_registry(fixture.runtime.workers).await; +} + +// 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"; + 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), + "a full worker-event backlog must not suppress terminal delivery maintenance" + ); + assert!( + 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; +} + +// 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, + ); + + 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"); +} + +// 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 +// 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, + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ); + 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 _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"); + // 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"); + 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, + &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" + ); + 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 +3399,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: Some("old transient failure".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3739,6 +4332,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3764,6 +4362,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3796,6 +4399,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: Some("previous blip".to_string()), withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3868,6 +4476,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -3905,6 +4518,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(), + &MessageInjectionMode::Wait, + delivery_retry_interval(), + ), last_error: None, withheld_fleet_ack: None, withheld_fleet_ack_floor: None, @@ -4324,6 +4942,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(), + &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 35fcae5787..55f31009f4 100644 --- a/crates/broker/src/runtime/util.rs +++ b/crates/broker/src/runtime/util.rs @@ -237,6 +237,33 @@ 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`]. +/// +/// 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. (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); + configured.clamp( + crate::broker::delivery_verification::VERIFICATION_WINDOW, + MAX_CONFIGURABLE_DELIVERY_AGE, + ) +} + // 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 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) } 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..d57e80f5a6 --- /dev/null +++ b/tests/relayflows/cases/1686-unacked-delivery-retries-forever/run.mjs @@ -0,0 +1,438 @@ +/** + * 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 three times the configured + * budget leaves ample scheduling margin. + */ +const HEAD_WINDOW_MS = 30_000; +/** + * 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 = 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 + * 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'); +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, { signal: AbortSignal.timeout(READINESS_TIMEOUT_MS) }); + 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', undefined, READINESS_TIMEOUT_MS).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 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 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 + // 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 = {}, 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(); + 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, 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(); + 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'); + // `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) { + 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'); +}