BrokerPort: cap the wait after a pending frame at a day

M3a review finding 8. The wait was expires plus the timeout with no bound,
and the "expiry too far away" guard could not fire, so a far expiry (which
brokerd produces when now + ttl_ms does not fit) parked a turn for ever.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

crates/loopd/src/broker_port.rs

@@ -19,6 +19,9 @@ pub const UNAVAILABLE: &str = "the tool broker is unavailable";
 pub const NOT_CONFIGURED: &str = "no tool broker is configured";
 /// The failure a request too big for one frame reaches.
 pub const TOO_LARGE: &str = "the request is too large for the tool broker";
+/// The longest `BrokerPort` waits for an answer after a pending frame, whatever its `expires`
+/// says, so a turn cannot be parked for ever by a far expiry.
+pub const MAX_PENDING_WAIT: Duration = Duration::from_secs(24 * 60 * 60);
 /// The runbook entry every outage line ends with.
 pub const POINTER: &str = "see docs/runbook.md#broker-unavailable";
 
@@ -55,6 +58,7 @@ impl std::io::Read for Deadline<'_> {
 pub struct BrokerPort {
     socket: PathBuf,
     timeout: Duration,
+    pending_cap: Duration,
     log: Box<dyn Fn(&str) + Send + Sync>,
 }
 
@@ -63,10 +67,19 @@ impl BrokerPort {
         BrokerPort {
             socket,
             timeout,
+            pending_cap: MAX_PENDING_WAIT,
             log: Box::new(|l| eprintln!("{l}")),
         }
     }
 
+    /// The same port with another cap on the wait after a pending frame (`MAX_PENDING_WAIT`).
+    pub fn with_pending_cap(self, pending_cap: Duration) -> BrokerPort {
+        BrokerPort {
+            pending_cap,
+            ..self
+        }
+    }
+
     pub fn with_log(
         socket: PathBuf,
         timeout: Duration,
@@ -75,6 +88,7 @@ impl BrokerPort {
         BrokerPort {
             socket,
             timeout,
+            pending_cap: MAX_PENDING_WAIT,
             log,
         }
     }
@@ -172,7 +186,8 @@ impl ToolPort for BrokerPort {
                     let wait = expires
                         .unix_millis()
                         .saturating_sub(Timestamp::now().unix_millis());
-                    until = match Instant::now().checked_add(Duration::from_millis(wait)) {
+                    let wait = Duration::from_millis(wait).min(self.pending_cap);
+                    until = match Instant::now().checked_add(wait) {
                         Some(until) => until,
                         None => return self.unavailable("an expiry too far away"),
                     };

crates/loopd/tests/broker_port_cap.rs

@@ -0,0 +1,60 @@
+//! After a pending frame, `BrokerPort` waits for `expires` plus its timeout, but never longer than
+//! its cap, whatever `expires` says (M3a review finding 8). `brokerd` falls back to the largest
+//! timestamp when `now + ttl_ms` does not fit, so a far `expires` is reachable by configuration.
+
+#[path = "support/broker.rs"]
+mod fake;
+
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use fake::{broker, request, send};
+use loopd::broker_port::{BrokerPort, MAX_PENDING_WAIT, UNAVAILABLE};
+use loopd::tools::ToolPort;
+use proto::{Timestamp, ToolResponse};
+
+#[test]
+fn the_default_cap_is_a_day() {
+    assert_eq!(MAX_PENDING_WAIT, Duration::from_secs(24 * 60 * 60));
+}
+
+#[test]
+fn a_pending_frame_with_the_largest_expiry_still_ends_at_the_cap() {
+    let (socket, broker) = broker(|stream, request| {
+        let pending = ToolResponse::PendingApproval {
+            approval: 7,
+            expires: Timestamp::MAX,
+        };
+        send(stream, request.id, false, pending);
+        thread::sleep(Duration::from_millis(1_500));
+    });
+    let lines = Arc::new(Mutex::new(Vec::new()));
+    let sink = lines.clone();
+    let port = BrokerPort::with_log(
+        socket,
+        Duration::from_millis(100),
+        Box::new(move |line| sink.lock().unwrap().push(line.to_string())),
+    )
+    .with_pending_cap(Duration::from_millis(200));
+
+    let started = Instant::now();
+    let answer = port.call(&request(), &mut |_| {});
+    let took = started.elapsed();
+    assert_eq!(
+        answer,
+        ToolResponse::Failed {
+            message: UNAVAILABLE.to_string()
+        }
+    );
+    assert!(
+        took >= Duration::from_millis(250),
+        "cap plus timeout: {took:?}"
+    );
+    assert!(
+        took < Duration::from_millis(1_200),
+        "gave up late: {took:?}"
+    );
+    assert_eq!(lines.lock().unwrap().len(), 1);
+    broker.join().unwrap();
+}