feat(cli): show pending message count in the pretty agent list

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `relay node agent list --pretty` shows a `PENDING` column with the messages still waiting to reach each agent (queued inbound plus in-flight deliveries awaiting confirmation); the broker reports it as `pending_messages` on `GET /api/spawned` and `GET /api/status`.
 - `agent-relay cloud login` now records your user, email, and organization to `~/.agentworkforce/relay/cloud-identity.json`, and the CLI, broker, and Relaycast traffic all report usage under that user and org instead of an anonymous machine id. `agent-relay cloud whoami` refreshes the record; `agent-relay cloud logout` clears it.
 - `agent-relay telemetry status` reports which user and organization usage is attributed to, or says so explicitly when it is anonymous.
 - Every event now carries a `machine_id` alongside the person key, signed in or not, so machine-level questions survive login: how many machines an account runs on, how many accounts share a machine, and (via Relaycast's `actor_machine_id`) how many machines share a workspace.

crates/broker/src/runtime/api.rs

@@ -1185,7 +1185,9 @@ impl BrokerRuntime {
                 );
             }
             ListenApiRequest::List { reply } => {
-                let _ = reply.send(Ok(json!({ "agents": workers.list() })));
+                let counts =
+                    super::delivery::pending_message_counts(delivery_states, pending_deliveries);
+                let _ = reply.send(Ok(json!({ "agents": workers.list(&counts) })));
             }
             ListenApiRequest::Threads { reply } => {
                 let mut messages: Vec<Value> = recent_thread_messages.iter().cloned().collect();
@@ -1648,7 +1650,10 @@ impl BrokerRuntime {
                     .collect();
                 let _ = reply.send(Ok(json!({
                     "agent_count": workers.workers.len(),
-                    "agents": workers.list(),
+                    "agents": workers.list(&super::delivery::pending_message_counts(
+                        delivery_states,
+                        pending_deliveries,
+                    )),
                     "pending_delivery_count": pending.len(),
                     "pending_deliveries": pending,
                     "node_connected": node_delivery_connected,

crates/broker/src/runtime/delivery.rs

@@ -421,6 +421,26 @@ pub(crate) struct InboundQueueResult {
     pub(crate) evicted_from: Option<String>,
 }
 
+/// Per-worker count of messages that have not reached the agent yet: the
+/// un-injected inbound queue (`manual_flush` backlog) plus the in-flight
+/// deliveries still awaiting worker confirmation. Feeds `pending_messages` on
+/// `GET /api/spawned` and `GET /api/status`; workers with nothing waiting are
+/// left out of the map.
+pub(crate) fn pending_message_counts(
+    delivery_states: &HashMap<WorkerName, InboundDeliveryState>,
+    pending_deliveries: &HashMap<DeliveryId, PendingDelivery>,
+) -> HashMap<WorkerName, usize> {
+    let mut counts: HashMap<WorkerName, usize> = delivery_states
+        .iter()
+        .filter(|(_, state)| state.pending_len() > 0)
+        .map(|(name, state)| (name.clone(), state.pending_len()))
+        .collect();
+    for delivery in pending_deliveries.values() {
+        *counts.entry(delivery.worker_name.clone()).or_insert(0) += 1;
+    }
+    counts
+}
+
 /// Build the `delivery_dropped` broker event for a queue-cap eviction.
 pub(crate) fn delivery_dropped_event_for_eviction(
     worker_name: &str,

crates/broker/src/runtime/tests.rs

@@ -42,15 +42,16 @@ use super::{
     is_unknown_worker_error_message, load_dead_letters, load_pending_deliveries,
     mark_delivery_read_ack, mark_delivery_read_ack_with_timeout, mint_or_recover_observer_token,
     normalize_channel, normalize_initial_task, normalize_sender, parse_sort_key_from_raw_timestamp,
-    persist_dead_letters_on_shutdown, persist_pending_on_shutdown, queue_inbound_for_delivery_mode,
-    relaycast_spawn_control_dedup_key, relaycast_ws_should_apply_local_spawn_echo_dedup,
-    relaycast_ws_spawn_token, requeue_dead_letter, resolve_exit_after_task, resolve_workspace,
-    retry_pending_delivery, save_dead_letters, seed_supplied_agent_token, send_broker_event,
-    sender_is_dashboard_label, should_clear_pending_delivery_for_event,
-    synthetic_delivery_read_ack_reason, AgentRuntime, DeadLetterEntry, DeadLetterStore,
-    DeliveryAttemptOutcome, InboundContext, InboundQueueOutcome, ObserverTokenMintError,
-    ObserverTokenMintOutcome, PendingDelivery, PendingDeliveryStore, ProtocolHeadlessProvider,
-    RelayWorkspace, TypedThreadMessage, MAX_DEAD_LETTERS, MAX_DELIVERY_RETRIES,
+    pending_message_counts, persist_dead_letters_on_shutdown, persist_pending_on_shutdown,
+    queue_inbound_for_delivery_mode, relaycast_spawn_control_dedup_key,
+    relaycast_ws_should_apply_local_spawn_echo_dedup, relaycast_ws_spawn_token,
+    requeue_dead_letter, resolve_exit_after_task, resolve_workspace, retry_pending_delivery,
+    save_dead_letters, seed_supplied_agent_token, send_broker_event, sender_is_dashboard_label,
+    should_clear_pending_delivery_for_event, synthetic_delivery_read_ack_reason, AgentRuntime,
+    DeadLetterEntry, DeadLetterStore, DeliveryAttemptOutcome, InboundContext, InboundQueueOutcome,
+    ObserverTokenMintError, ObserverTokenMintOutcome, PendingDelivery, PendingDeliveryStore,
+    ProtocolHeadlessProvider, RelayWorkspace, TypedThreadMessage, MAX_DEAD_LETTERS,
+    MAX_DELIVERY_RETRIES,
 };
 use crate::dedup::DedupCache;
 use crate::relaycast::{
@@ -266,6 +267,48 @@ async fn inbound_queue_manual_flush_holds_until_explicit_drain() {
     cleanup_worker_registry(workers).await;
 }
 
+#[tokio::test]
+async fn worker_list_reports_pending_queue_depth() {
+    let worker_name = "worker-a";
+    let workers = make_worker_registry_with_worker(worker_name).await;
+    let mut delivery_states = HashMap::from([(
+        WorkerName::from(worker_name),
+        InboundDeliveryState::new(InboundDeliveryMode::ManualFlush),
+    )]);
+
+    let mut pending_deliveries = HashMap::new();
+
+    let counts = pending_message_counts(&delivery_states, &pending_deliveries);
+    assert_eq!(workers.list(&counts)[0]["pending_messages"], 0);
+
+    for event_id in ["evt_1", "evt_2"] {
+        queue_inbound_for_delivery_mode(
+            &mut delivery_states,
+            &workers,
+            worker_name,
+            inbound_ctx(event_id),
+        );
+    }
+    pending_deliveries.insert(
+        DeliveryId::new("del_in_flight"),
+        pending_delivery(worker_name, "del_in_flight", "evt_3"),
+    );
+
+    let counts = pending_message_counts(&delivery_states, &pending_deliveries);
+    assert_eq!(
+        workers.list(&counts)[0]["pending_messages"],
+        3,
+        "queued inbound messages plus in-flight deliveries are both still pending"
+    );
+    assert_eq!(
+        workers.list(&HashMap::new())[0]["pending_messages"],
+        0,
+        "a worker with neither queue populated has nothing pending"
+    );
+
+    cleanup_worker_registry(workers).await;
+}
+
 #[tokio::test]
 async fn inbound_queue_worker_missing_does_not_create_state() {
     let (tx, _rx) = mpsc::channel::<WorkerEvent>(16);

crates/broker/src/types.rs

@@ -336,6 +336,11 @@ impl InboundDeliveryState {
     pub fn pending_snapshot(&self) -> Vec<PendingRelayMessage> {
         self.pending.iter().cloned().collect()
     }
+
+    /// Number of queued inbound messages waiting to reach the worker.
+    pub fn pending_len(&self) -> usize {
+        self.pending.len()
+    }
 }
 
 #[cfg(test)]

crates/broker/src/worker.rs

@@ -139,7 +139,11 @@ impl WorkerRegistry {
         Some(self.worker_logs_dir.join(format!("{worker_name}.log")))
     }
 
-    pub(crate) fn list(&self) -> Vec<Value> {
+    /// Snapshot every spawned worker for `GET /api/spawned` and `GET /api/status`.
+    ///
+    /// `pending_messages` comes from [`crate::runtime::pending_message_counts`];
+    /// a worker missing from the map has nothing waiting.
+    pub(crate) fn list(&self, pending_messages: &HashMap<WorkerName, usize>) -> Vec<Value> {
         self.workers
             .iter()
             .map(|(name, handle)| {
@@ -162,6 +166,7 @@ impl WorkerRegistry {
                         - chrono::Duration::from_std(handle.last_activity_at.elapsed()).unwrap_or_default(),
                     "context_budget_pct": handle.context_budget_pct,
                     "current_state": handle.state.as_str(),
+                    "pending_messages": pending_messages.get(name).copied().unwrap_or(0),
                     "runtime_kind": if native_harness.is_some() { "native" } else if handle.spec.runtime == AgentRuntime::Pty { "pty" } else { "headless" },
                     "native_harness_protocol_version": native_harness.as_ref().map(|(version, _)| *version),
                     "native_harness_capabilities": native_harness.and_then(|(_, capabilities)| capabilities),
@@ -1890,7 +1895,7 @@ mod tests {
     #[test]
     fn worker_registry_starts_empty() {
         let reg = make_registry(vec![]);
-        assert!(reg.list().is_empty());
+        assert!(reg.list(&HashMap::new()).is_empty());
     }
 
     #[test]

learnings/duplicated-wire-contract.md

@@ -0,0 +1,101 @@
+# Duplicated wire contract
+
+**One producer serializes one payload; the consumer describes it twice by hand.
+Adding a field updates one copy. Nothing breaks. The other copy is now a lie.**
+
+## The shape of it
+
+A producer emits a payload from a single place — one function, one serializer,
+one query. Consumers in another language (or another module) hand-write a type,
+schema, or struct to describe it. Then a _second_ consumer surface needs the
+same payload, and someone hand-writes a second description instead of pointing
+at the first.
+
+Now there are two descriptions of one runtime shape, with no mechanical link
+between them. Every future field addition is a coin flip over which copies get
+updated.
+
+## Why it survives review
+
+Nothing fails. That is the whole problem.
+
+- **The compiler is satisfied.** Both declarations are internally valid. A type
+  that omits a field the JSON actually carries is not an error — it is just a
+  narrower view.
+- **The tests pass.** The runtime data is correct. Only the _description_ of it
+  is wrong, and descriptions are not exercised at runtime.
+- **The failure is asymmetric.** Consumers coming through the updated door see
+  the field and are happy. Only consumers coming through the stale door hit the
+  wall — and they hit it later, in different code, far from the change that
+  caused it.
+- **Drift accumulates silently.** Nobody notices a missing field until someone
+  needs it. By then several are gone, and the gap looks like intent rather than
+  rot.
+
+## Why the obvious fix makes it worse
+
+The natural response to "field missing from copy B" is to add the field to
+copy B. That resolves the symptom and preserves the cause: two hand-maintained
+copies, still unlinked, now with a fresh precedent that keeping them in sync by
+hand is the way this codebase works. The next field will drift too.
+
+**Repair the link, not the copy.** Delete the second declaration and point it at
+the first. If the two genuinely cannot be unified — different ownership,
+different modules, a real boundary — add a compile-time assertion that one is
+assignable to the other, so the next drift fails a build instead of a user.
+
+## How to catch it before you ship
+
+When adding a field to a hand-written type that mirrors an external producer:
+
+**Grep for a sibling field, not the new one.** The new field is what you are
+adding; it appears nowhere else by definition. Pick a distinctive field that has
+been on the payload for a while and search for it. Two hits in two files means
+two declarations of one payload.
+
+If the second hit is an inline anonymous type nested inside a larger response
+type, treat that as the strongest signal — an inline shape has no name to search
+for, so it is the copy that gets forgotten.
+
+## Where it breeds
+
+- Cross-language boundaries. A Rust producer and a TypeScript consumer share no
+  compiler, so the only thing keeping them aligned is attention.
+- Anywhere one payload is reachable through more than one endpoint, method, or
+  response envelope. Multiple doors to one room invite one description per door.
+- Generated-code gaps. If part of a contract is generated and part is
+  hand-written, the hand-written part is where this lives.
+
+## What it is not
+
+This is not general code duplication, and "DRY" is not the useful frame. Two
+similar-looking types that describe two genuinely different payloads should stay
+separate; unifying those couples things that are free to diverge. The defect is
+specific: **two descriptions of one runtime shape**, where only one description
+is ever checked against reality.
+
+## Worked example (relay, PR #1365)
+
+`GET /api/spawned` and the `agents` array of `GET /api/status` are both
+serialized from a single `WorkerRegistry::list` call in the broker. TypeScript
+described that payload twice:
+
+- `ListAgent` in `packages/harness-driver/src/types.ts` — used by `listAgents()`
+- an inline anonymous object type inside `BrokerStatus.agents` in
+  `packages/harness-driver/src/protocol.ts` — used by `getStatus()`
+
+Adding `pending_messages` to `ListAgent` left `getStatus()` consumers unable to
+reach a field the broker was demonstrably sending. Investigating showed the copy
+had _already_ drifted by four fields — `sessionId`, `runtime_kind`,
+`native_harness_protocol_version`, `native_harness_capabilities` — none of which
+anyone had noticed.
+
+Caught by a review bot. Not by the compiler, not by the tests, not by the author.
+That is the tell that this class needs a structural fix rather than more care.
+
+Fixed by moving `ListAgent` into the wire-contract module next to `BrokerStatus`
+and pointing `BrokerStatus.agents` at it, so there is one declaration.
+`types.ts` re-exports it, keeping every existing import path valid.
+
+Still outstanding: the broker also emits `workerPid`, which the unified type
+does not declare — the same drift, one layer down.

packages/cli/src/cli/commands/local-agent.test.ts

@@ -98,6 +98,7 @@ describe('local agent subtree', () => {
                 model: 'gpt-5.4',
                 channels: [],
                 current_state: 'working' as const,
+                pending_messages: 3,
                 last_activity_at: '2026-07-22T16:59:57.000Z',
               },
             ]),
@@ -108,7 +109,9 @@ describe('local agent subtree', () => {
 
     await program.parseAsync(['local', 'agent', 'list', '--pretty'], { from: 'user' });
 
-    expect(log).toHaveBeenCalledWith(expect.stringContaining('Lead  codex / gpt-5.4  ● working  now'));
+    expect(log).toHaveBeenCalledWith(
+      expect.stringContaining('Lead  codex / gpt-5.4  ● working  3        now')
+    );
   });
 
   it('formats agent state and activity time for the pretty list', () => {
@@ -121,16 +124,37 @@ describe('local agent subtree', () => {
             cli: 'claude',
             channels: [],
             current_state: 'blocked_on_send',
+            pending_messages: 2,
             last_activity_at: '2026-07-22T16:58:00.000Z',
           },
           { name: 'Worker', runtime: 'headless', channels: [], current_state: 'idle' },
         ],
         new Date('2026-07-22T17:00:00.000Z')
       )
-    ).toMatch(/Review\s+claude\s+◐ waiting\s+2 minutes ago/);
+    ).toMatch(/Review\s+claude\s+◐ waiting\s+2\s+2 minutes ago/);
     expect(formatPrettyAgentList([], new Date())).toBe('No agents running.');
   });
 
+  it('shows the pending queue depth, and "-" when the broker omits it', () => {
+    const [header, , withCount, withoutCount] = formatPrettyAgentList(
+      [
+        {
+          name: 'Review',
+          runtime: 'pty',
+          channels: [],
+          current_state: 'blocked_on_send',
+          pending_messages: 4,
+        },
+        { name: 'Worker', runtime: 'headless', channels: [], current_state: 'idle' },
+      ],
+      new Date('2026-07-22T17:00:00.000Z')
+    ).split('\n');
+
+    expect(header).toMatch(/STATE\s+PENDING\s+LAST ACTIVE/);
+    expect(withCount).toMatch(/◐ waiting\s+4\s+unknown/);
+    expect(withoutCount).toMatch(/○ idle\s+-\s+unknown/);
+  });
+
   it('removes terminal control sequences from broker-provided cells', () => {
     const [, , row] = formatPrettyAgentList(
       [

packages/cli/src/cli/commands/local-agent.ts

@@ -151,6 +151,16 @@ function formatRelativeTime(value: string | undefined, now: Date): string {
   return `${days} day${days === 1 ? '' : 's'} ago`;
 }
 
+/**
+ * Render the count of messages still waiting to reach the agent. Brokers older
+ * than the field report nothing, which stays `-` rather than claiming an empty
+ * queue.
+ */
+function formatPendingCount(value: number | undefined): string {
+  if (typeof value !== 'number' || !Number.isFinite(value) || value < 0) return '-';
+  return String(Math.floor(value));
+}
+
 /** Keep broker-provided text from escaping its table cell or controlling the terminal. */
 function sanitizeTerminalCell(value: string): string {
   // eslint-disable-next-line no-control-regex
@@ -169,13 +179,15 @@ export function formatPrettyAgentList(agents: ListAgent[], now: Date): string {
         [agent.cli ?? agent.provider ?? agent.runtime, agent.model].filter(Boolean).join(' / ')
       ),
       state: sanitizeTerminalCell(state ? `${state.symbol} ${state.label}` : '· unknown'),
+      pending: formatPendingCount(agent.pending_messages),
       lastActive: sanitizeTerminalCell(formatRelativeTime(agent.last_activity_at, now)),
     };
   });
   const columns = [
     { header: 'NAME', values: rows.map((row) => row.name) },
     { header: 'CLI / MODEL', values: rows.map((row) => row.cliModel) },
     { header: 'STATE', values: rows.map((row) => row.state) },
+    { header: 'PENDING', values: rows.map((row) => row.pending) },
     { header: 'LAST ACTIVE', values: rows.map((row) => row.lastActive) },
   ];
   const widths = columns.map((column) =>
@@ -190,7 +202,7 @@ export function formatPrettyAgentList(agents: ListAgent[], now: Date): string {
   return [
     formatRow(columns.map((column) => column.header)),
     formatRow(columns.map((_, index) => '-'.repeat(widths[index]!))),
-    ...rows.map((row) => formatRow([row.name, row.cliModel, row.state, row.lastActive])),
+    ...rows.map((row) => formatRow([row.name, row.cliModel, row.state, row.pending, row.lastActive])),
   ].join('\n');
 }