Merge pull request #1364 from AgentWorkforce/claude/relay-e2e-tic-tac-toe-hfjyvv

fix(cli,broker): unbreak the attach panes — clamp the status line, reap dead agents

Author: willwashburn

CHANGELOG.md

@@ -28,6 +28,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - The published CLI now actually reports telemetry. The npm package is plain `tsc` output with no key injection step, and the bun standalone's `--define` targeted a literal `process.env.AGENT_RELAY_POSTHOG_KEY` that the code never read (it used a computed `process.env[name]` lookup), so **both** installable artifacts shipped with telemetry silently disabled — every `cli_command_run`, `workflow_run`, `cloud_auth`, `agent_relay_tool_call`, `setup_init`, `swarm_run`, and `bridge_spawn` event was dropped. Only the Rust broker was reporting.
 - Opting out of telemetry (`AGENT_RELAY_TELEMETRY_DISABLED` or `DO_NOT_TRACK`) now keeps your cloud identity out of child process environments, including identity an ancestor process or your shell had already exported. The identity env vars — one of which carries your email — previously reached every spawned process, including third-party harness CLIs, even when opted out.
 - Identity forwarding to the Relaycast gateway is no longer gated on the local process carrying a PostHog key. An npm-installed CLI bakes no key, so it previously forwarded no identity at all and every hosted event fell back to being keyed on the workspace. Forwarding now follows the telemetry preference alone.
+- `node agent list` no longer reports an exited agent as `working` indefinitely; the broker now reaps a worker whose harness has gone away or never started.
+- `node agent attach --mode drive|passthrough` truncates its status line to the terminal width, so a narrow pane no longer stacks status lines over the agent's output.
 - `node agent attach --mode view` now exits on the first Ctrl-C instead of waiting for a WebSocket close handshake.
 - The broker now sends its anonymous telemetry id (`X-Agent-Relay-Distinct-Id`) and origin actor with its Relaycast requests, so hosted usage can be attributed to an install instead of only to a workspace. The id header is omitted when telemetry is opted out; requests and origin actor are unaffected.
 - The broker now reads its telemetry preference and machine-id files from `AGENT_RELAY_DATA_DIR` when set, matching the CLI. It previously only read `~/.agentworkforce/relay/telemetry.json`, so an opt-out written by `agent-relay telemetry disable` under a configured data directory was ignored.

crates/broker/src/runtime/tests.rs

@@ -107,6 +107,9 @@ async fn make_worker_registry_with_worker(name: &str) -> WorkerRegistry {
             stdin,
             harness_pid: None,
             spawned_at: Instant::now(),
+            // Ready, so the orphan sweep's readiness deadline never applies to
+            // these fixtures.
+            ready_at: Some(Instant::now()),
             last_activity_at: Instant::now(),
             context_budget_pct: None,
             state: AgentWorkState::Working,

crates/broker/src/runtime/worker_events.rs

@@ -848,6 +848,10 @@ impl BrokerRuntime {
                                 if let Some(pid) = payload_pid {
                                     h.harness_pid = Some(pid);
                                 }
+                                // Records that the harness actually came up, so
+                                // `reap_exited` can tell a slow start from one that
+                                // never happened.
+                                h.ready_at.get_or_insert_with(Instant::now);
                                 (
                                     h.spec.provider.clone(),
                                     h.spec.cli.clone(),

crates/broker/src/worker.rs

@@ -42,6 +42,83 @@ const APP_SERVER_AUTH_ENV_KEYS: [&str; 4] = [
 const DEFAULT_RELEASE_GRACE: Duration = Duration::from_secs(2);
 const APP_SERVER_RELEASE_GRACE: Duration = Duration::from_secs(35);
 
+/// How long a worker may go without reporting `worker_ready` before the broker
+/// treats its harness as failed-to-start.
+///
+/// The worker process emits `worker_ready` itself — the PTY runtime even has a
+/// 25s fallback that fires when readiness detection times out
+/// (`pty_worker::STARTUP_READY_TIMEOUT`). So silence past this deadline does not
+/// mean "slow": it means the worker died, or wedged, before it could report.
+/// The margin over that 25s fallback is deliberately generous — reaping a
+/// healthy-but-slow agent is far worse than listing a dead one a little longer.
+const WORKER_READY_DEADLINE: Duration = Duration::from_secs(90);
+
+/// How long to wait for a SIGKILLed orphan wrapper to be reaped before giving
+/// up. Bounded so a wrapper stuck in uninterruptible sleep cannot stall the
+/// maintenance tick, which also drives delivery retries.
+const ORPHAN_REAP_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// Why a worker was reaped despite its wrapper process still being alive.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum OrphanedWorker {
+    /// The harness pid the worker reported is gone.
+    HarnessExited,
+    /// `worker_ready` never arrived within [`WORKER_READY_DEADLINE`].
+    NeverReady,
+}
+
+impl OrphanedWorker {
+    fn reason(self) -> &'static str {
+        match self {
+            OrphanedWorker::HarnessExited => "harness_exited",
+            OrphanedWorker::NeverReady => "harness_never_ready",
+        }
+    }
+}
+
+/// True when `pid` no longer names a live process.
+///
+/// Safety: `kill(pid, 0)` is a POSIX-safe probe — it performs the permission
+/// and existence checks without delivering a signal. `ESRCH` means the process
+/// is gone; every other error (notably `EPERM`) means it exists.
+#[cfg(unix)]
+fn pid_is_gone(pid: u32) -> bool {
+    let ret = unsafe { libc::kill(pid as libc::pid_t, 0) };
+    ret == -1 && std::io::Error::last_os_error().raw_os_error().unwrap_or(0) == libc::ESRCH
+}
+
+#[cfg(not(unix))]
+fn pid_is_gone(_pid: u32) -> bool {
+    false
+}
+
+/// Decide whether a worker is dead even though its wrapper process is alive.
+///
+/// The wrapper (`agent-relay-broker pty …`) can outlive the harness it hosts:
+/// its stdin reader blocks on a pipe the broker never closes, so a wrapper whose
+/// harness exited at startup can sit in `futex_do_wait` indefinitely. Reaping on
+/// the wrapper alone therefore leaves the agent listed as `working` forever, with
+/// `last_activity` frozen at the moment it died. Judge liveness by the harness.
+pub(crate) fn orphaned_worker(
+    harness_pid: Option<u32>,
+    ready_at: Option<Instant>,
+    spawned_at: Instant,
+    now: Instant,
+) -> Option<OrphanedWorker> {
+    if let Some(pid) = harness_pid {
+        if pid_is_gone(pid) {
+            return Some(OrphanedWorker::HarnessExited);
+        }
+        // A live harness pid is proof of life; never apply the readiness
+        // deadline to one that is plainly running.
+        return None;
+    }
+    if ready_at.is_none() && now.saturating_duration_since(spawned_at) > WORKER_READY_DEADLINE {
+        return Some(OrphanedWorker::NeverReady);
+    }
+    None
+}
+
 // Working/idle activity inference from PTY output comes from the
 // harness-agnostic `relay-pty` crate.
 pub(crate) use relay_pty::detection;
@@ -55,6 +132,9 @@ pub(crate) struct WorkerHandle {
     pub(crate) stdin: ChildStdin,
     pub(crate) harness_pid: Option<u32>,
     pub(crate) spawned_at: Instant,
+    /// When the worker reported `worker_ready`. `None` means the harness has
+    /// never come up — see `WORKER_READY_DEADLINE` in `reap_exited`.
+    pub(crate) ready_at: Option<Instant>,
     pub(crate) last_activity_at: Instant,
     pub(crate) context_budget_pct: Option<u8>,
     pub(crate) state: AgentWorkState,
@@ -232,16 +312,7 @@ impl WorkerRegistry {
         #[cfg(unix)]
         {
             match handle.child.id() {
-                // Safety: kill(pid, 0) is a POSIX-safe probe that checks process
-                // existence without sending a signal. ESRCH => the process is gone.
-                Some(pid) => {
-                    let ret = unsafe { libc::kill(pid as libc::pid_t, 0) };
-                    if ret == -1 {
-                        std::io::Error::last_os_error().raw_os_error().unwrap_or(0) != libc::ESRCH
-                    } else {
-                        true
-                    }
-                }
+                Some(pid) => !pid_is_gone(pid),
                 // `id()` returns None once the child has been waited/reaped.
                 None => false,
             }
@@ -867,6 +938,7 @@ impl WorkerRegistry {
             stdin,
             harness_pid: initial_harness_pid,
             spawned_at: Instant::now(),
+            ready_at: None,
             last_activity_at: Instant::now(),
             context_budget_pct: None,
             state: AgentWorkState::Working,
@@ -1003,24 +1075,13 @@ impl WorkerRegistry {
                             #[cfg(unix)]
                             {
                                 if let Some(pid) = handle.child.id() {
-                                    // Safety: kill(pid, 0) is a POSIX-safe probe that checks
-                                    // process existence without sending a signal. ESRCH means
-                                    // the process no longer exists.
-                                    let ret = unsafe { libc::kill(pid as libc::pid_t, 0) };
-                                    if ret == -1 {
-                                        let errno = std::io::Error::last_os_error()
-                                            .raw_os_error()
-                                            .unwrap_or(0);
-                                        if errno == libc::ESRCH {
-                                            tracing::info!(
-                                                worker = %name,
-                                                pid = pid,
-                                                "reap_exited: kill(0) says ESRCH — process gone"
-                                            );
-                                            (None, true)
-                                        } else {
-                                            (None, false)
-                                        }
+                                    if pid_is_gone(pid) {
+                                        tracing::info!(
+                                            worker = %name,
+                                            pid = pid,
+                                            "reap_exited: kill(0) says ESRCH — process gone"
+                                        );
+                                        (None, true)
                                     } else {
                                         (None, false)
                                     }
@@ -1048,6 +1109,62 @@ impl WorkerRegistry {
             } else {
                 (None, false)
             };
+            // The wrapper can outlive its harness. When it does, judge the agent
+            // by the harness and tear the orphaned wrapper down — otherwise the
+            // agent is listed as `working` forever.
+            let orphaned = if status.is_none() && !gone_via_kill0 {
+                self.workers.get(&name).and_then(|handle| {
+                    orphaned_worker(
+                        handle.harness_pid,
+                        handle.ready_at,
+                        handle.spawned_at,
+                        Instant::now(),
+                    )
+                })
+            } else {
+                None
+            };
+            if let Some(orphan) = orphaned {
+                let reason = self
+                    .workers
+                    .get(&name)
+                    .and_then(|handle| handle.exit_reason.clone())
+                    .or_else(|| Some(orphan.reason().to_string()));
+                if let Some(handle) = self.workers.get_mut(&name) {
+                    tracing::warn!(
+                        worker = %name,
+                        wrapper_pid = ?handle.child.id(),
+                        harness_pid = ?handle.harness_pid,
+                        reason = orphan.reason(),
+                        "reap_exited: harness gone but wrapper alive — killing orphaned wrapper"
+                    );
+                    // SIGKILL *and* reap. Dropping the `Child` without waiting
+                    // leaves the wrapper to tokio's best-effort background
+                    // reaper, so a run of failed agent starts accumulates
+                    // zombies. SIGKILL cannot be caught, so this returns
+                    // promptly — but the deadline keeps a wrapper wedged in
+                    // uninterruptible sleep from stalling the maintenance tick,
+                    // which also drives delivery retries.
+                    if let Err(error) = handle.child.start_kill() {
+                        tracing::warn!(worker = %name, %error, "failed to signal orphaned wrapper");
+                    }
+                    match timeout(ORPHAN_REAP_TIMEOUT, handle.child.wait()).await {
+                        Ok(Ok(_)) => {}
+                        Ok(Err(error)) => {
+                            tracing::warn!(worker = %name, %error, "orphaned wrapper wait failed")
+                        }
+                        Err(_) => tracing::warn!(
+                            worker = %name,
+                            timeout_ms = ORPHAN_REAP_TIMEOUT.as_millis(),
+                            "orphaned wrapper did not exit before the reap deadline"
+                        ),
+                    }
+                }
+                self.workers.remove(&name);
+                self.initial_tasks.remove(&name);
+                exited.push((name, None, None, reason));
+                continue;
+            }
             if let Some(status) = status {
                 let code = status.code();
                 #[cfg(unix)]
@@ -1898,6 +2015,86 @@ mod tests {
         assert!(reg.list(&HashMap::new()).is_empty());
     }
 
+    // The wrapper process can outlive the harness it hosts, so reaping on the
+    // wrapper alone leaves a dead agent listed as `working` forever.
+    mod orphaned_worker {
+        use super::*;
+
+        /// A pid that cannot be live. `kill(0)` on pid 0 addresses the caller's
+        /// own process group, so use an unassigned high pid instead.
+        fn dead_pid() -> u32 {
+            // Above the default pid_max; never allocated.
+            0x7FFF_FFFF
+        }
+
+        fn live_pid() -> u32 {
+            std::process::id()
+        }
+
+        #[test]
+        fn reports_harness_exited_when_the_reported_pid_is_gone() {
+            let now = Instant::now();
+            assert_eq!(
+                orphaned_worker(Some(dead_pid()), Some(now), now, now),
+                Some(OrphanedWorker::HarnessExited)
+            );
+        }
+
+        #[test]
+        fn leaves_a_worker_with_a_live_harness_alone() {
+            let now = Instant::now();
+            assert_eq!(orphaned_worker(Some(live_pid()), Some(now), now, now), None);
+        }
+
+        // A live harness pid is proof of life even if `worker_ready` was missed,
+        // so the readiness deadline must not apply to it.
+        #[test]
+        fn a_live_harness_is_never_reaped_for_missing_readiness() {
+            let spawned = Instant::now();
+            let now = spawned + WORKER_READY_DEADLINE + Duration::from_secs(60);
+            assert_eq!(orphaned_worker(Some(live_pid()), None, spawned, now), None);
+        }
+
+        #[test]
+        fn reports_never_ready_only_after_the_deadline() {
+            let spawned = Instant::now();
+            // Still inside the window — a slow harness must be left to boot.
+            assert_eq!(
+                orphaned_worker(None, None, spawned, spawned + Duration::from_secs(30)),
+                None
+            );
+            assert_eq!(
+                orphaned_worker(None, None, spawned, spawned + WORKER_READY_DEADLINE),
+                None
+            );
+            assert_eq!(
+                orphaned_worker(
+                    None,
+                    None,
+                    spawned,
+                    spawned + WORKER_READY_DEADLINE + Duration::from_secs(1)
+                ),
+                Some(OrphanedWorker::NeverReady)
+            );
+        }
+
+        // A worker that reported ready and has no pid to probe (non-PTY
+        // runtimes) must never be reaped by the deadline.
+        #[test]
+        fn a_ready_worker_without_a_pid_is_never_reaped() {
+            let spawned = Instant::now();
+            let now = spawned + WORKER_READY_DEADLINE * 10;
+            assert_eq!(orphaned_worker(None, Some(spawned), spawned, now), None);
+        }
+
+        #[test]
+        fn the_deadline_clears_the_pty_runtime_startup_fallback() {
+            // `pty_worker::STARTUP_READY_TIMEOUT` is 25s; the broker must wait
+            // comfortably longer than the worker's own fallback.
+            assert!(WORKER_READY_DEADLINE > Duration::from_secs(25) * 3);
+        }
+    }
+
     #[test]
     fn has_worker_returns_false_for_unknown() {
         let reg = make_registry(vec![]);