fix(broker): persist correlated worker exits

.github/pull_request_template.md

@@ -13,8 +13,8 @@ Replace both values below. Use `feature` or `bugfix` for user-visible behavior
 changes and add exactly one case under `tests/relayflows/cases/<case-id>/`.
 Use `non-functional` and `n/a` only when runtime behavior is unchanged.
 
-- Change type: `replace-me` <!-- relay-pr-proof:type -->
-- RelayFlow case: `replace-me` <!-- relay-pr-proof:case -->
+- Change type: `feature`, `bugfix`, or `non-functional` <!-- relay-pr-proof:type -->
+- RelayFlow case: `<exact case id>` or `n/a` for non-functional changes <!-- relay-pr-proof:case -->
 
 ## Screenshots
 

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `agent-relay node up` retries the narrowly transient Relaycast `workspace_busy` admission response while keeping unrelated rate limits terminal and preserving bounded startup diagnostics.
 
+- Worker deaths now persist bounded, generation-correlated exit diagnostics immediately and surface them through node and Fleet agent events. Fleet invocation correlation is carried on the exited process generation itself, so a same-name replacement worker can no longer overwrite the exit attribution. Hosted `agent_exited` delivery is now a durable, at-least-once outbox tied to the persisted crash record: a full or closed hosted-event channel no longer silently drops the terminal event, pending deliveries survive a broker restart and are replayed in order (deduped by agent + generation), and pending/backlog/drop counts are queryable via the crash-insights API. The durable record is only marked delivered after Relaycast's HTTP call actually succeeds (with a bounded in-process retry/backoff on timeout or 5xx) — not merely after the event reached the internal publisher queue — so a Relaycast outage no longer permanently loses an exit event; the crash-insights API also reports a `publish_failures_total` count so a stuck outage is observable rather than silently retried forever. A failed write of the crash-insights snapshot itself is now retried automatically on every subsequent maintenance flush (instead of only being logged), with `save_failures_total`/`dirty` and a `retention_pressure_total` counter exposed via the crash-insights API so a transient disk failure or sustained pending-outbox pressure is operator-visible rather than silently lost. The in-memory hosted-delivery queue is now continuously replenished from the durable pending outbox after it overflows its 256-entry bound, within the running broker process — no restart required — without ever re-queuing a delivery that is already in flight.
+
 - `fleet spawn --sandbox` uses the provider-neutral durable profile for explicit Daytona and E2B sandboxes, so their measured resource envelopes are routable while Agent37 retains its heavy profile.
 
 - Cloud Daytona Fleet provisioning now requires and returns the exact provider sandbox UUID alongside the stable Cloud sandbox ID, enabling ID-bound inspection and cleanup after interrupted launches.

crates/broker/src/runtime/api.rs

@@ -314,6 +314,10 @@ impl BrokerRuntime {
         let persist = self.persist;
         let shutdown = &mut self.shutdown;
         let crash_insights = &self.crash_insights;
+        let hosted_agent_exit_backlog_len = self.hosted_agent_exit_backlog.len();
+        let hosted_agent_exit_dropped_total = self.hosted_agent_exit_dropped_total;
+        let hosted_agent_exit_publish_failures_total =
+            self.hosted_agent_exit_publish_failures_total;
 
         match req {
             ListenApiRequest::Spawn {
@@ -674,6 +678,13 @@ impl BrokerRuntime {
                         }
                         if let Some((token, invocation_id, session_ref)) = fleet_registration.take()
                         {
+                            // Carry correlation on this specific generation's
+                            // handle, not just the by-name `fleet_inventory`
+                            // entry: a same-name replacement can overwrite that
+                            // entry before this generation is reaped, which
+                            // would misattribute this invocation id to the
+                            // wrong exit. See maintenance.rs reap.
+                            workers.set_invocation_id(&name, invocation_id.clone());
                             super::fleet::record_fleet_inventory_agent(
                                 fleet_control_tx,
                                 fleet_inventory,
@@ -2077,7 +2088,35 @@ impl BrokerRuntime {
                 })));
             }
             ListenApiRequest::GetCrashInsights { reply } => {
-                let _ = reply.send(Ok(crash_insights.to_json()));
+                // Backward-compatible additions: existing fields from
+                // `crash_insights.to_json()` are unchanged; `hosted_delivery`
+                // groups the durable outbox/backlog status a REST client can
+                // poll instead of (or alongside) directly reading
+                // `crash-insights.json`'s per-record `hosted_delivery` state.
+                let mut body = crash_insights.to_json();
+                if let Some(object) = body.as_object_mut() {
+                    object.insert(
+                        "hosted_delivery".to_string(),
+                        json!({
+                            "pending": crash_insights.pending_hosted_deliveries().len(),
+                            "in_memory_backlog_len": hosted_agent_exit_backlog_len,
+                            "in_memory_backlog_cap": super::event_loop::HOSTED_AGENT_EXIT_BACKLOG_CAP,
+                            "dropped_total": hosted_agent_exit_dropped_total,
+                            // Truthful operator status: a record only ever
+                            // counts as `pending` above until Relaycast's
+                            // real HTTP emit is confirmed by
+                            // `run_hosted_agent_event_publisher` — never
+                            // merely because it reached that task's queue.
+                            // `publish_failures_total` counts hosted events
+                            // whose bounded in-process retries were
+                            // exhausted without success; those records
+                            // remain `pending` above and are candidates for
+                            // restart replay, not silently marked delivered.
+                            "publish_failures_total": hosted_agent_exit_publish_failures_total,
+                        }),
+                    );
+                }
+                let _ = reply.send(Ok(body));
             }
             ListenApiRequest::GetDeadLetters { reply } => {
                 let now_ms = unix_timestamp_millis();

crates/broker/src/runtime/delivery.rs

@@ -970,6 +970,17 @@ pub(crate) async fn retry_pending_delivery(
     {
         Ok(()) => {
             if let Some(current) = pending_deliveries.get_mut(delivery_id) {
+                // `attempts` is a pure observability counter of successful
+                // handoffs, never the failure-budget gate — only
+                // `failed_attempts` (reset below) is compared against
+                // `MAX_DELIVERY_RETRIES` to decide termination. A long-lived
+                // `Wait`-mode delivery can be handed off successfully far
+                // more than `MAX_DELIVERY_RETRIES` times while its ack is
+                // still outstanding (each successful handoff resets the
+                // failure budget), so capping this counter at the failure
+                // budget previously misrepresented — and, if ever read as a
+                // budget check elsewhere, could prematurely treat — a
+                // perfectly healthy, still-pending delivery as exhausted.
                 current.attempts = current.attempts.saturating_add(1);
                 current.failed_attempts = 0;
                 current.next_retry_at = Instant::now()
@@ -985,8 +996,11 @@ pub(crate) async fn retry_pending_delivery(
         }
         Err(error) => {
             let should_fail = if let Some(current) = pending_deliveries.get_mut(delivery_id) {
-                current.attempts = current.attempts.saturating_add(1);
-                current.failed_attempts = current.failed_attempts.saturating_add(1);
+                current.attempts = current.attempts.saturating_add(1).min(MAX_DELIVERY_RETRIES);
+                current.failed_attempts = current
+                    .failed_attempts
+                    .saturating_add(1)
+                    .min(MAX_DELIVERY_RETRIES);
                 current.next_retry_at = Instant::now() + retry_interval;
                 current.last_error = Some(error.to_string());
                 current.failed_attempts >= MAX_DELIVERY_RETRIES