Fix/stale summary jobs

README.md

@@ -240,6 +240,10 @@ Pull requests are welcome. Before opening one:
 3. Follow the commit convention: `feat: added …` / `fix: corrected …`
 4. Branch names: `feat/short-description` or `fix/short-description`
 
+About to touch the memory or bundling code? [`docs/pendiente.md`](docs/pendiente.md)
+records what was open, how each point was closed and — just as usefully — what
+was looked at and deliberately left alone, so nobody relitigates it.
+
 ---
 
 ## License

daemon/bento-cli/Cargo.toml

@@ -15,6 +15,7 @@ crossterm = { version = "0.29", features = ["event-stream"] }
 tokio-stream = { version = "0.1", default-features = false }
 bento-review = { path = "../bento-review" }
 bento-sessions = { path = "../bento-sessions" }
+vt100 = "0.16.2"
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"

daemon/bento-cli/src/main.rs

@@ -134,8 +134,34 @@ pub(crate) async fn stream_review(body: Value) -> std::io::Result<()> {
     Ok(())
 }
 
+/// How long a single request may take before the caller gives up. The panel
+/// awaits these on its event loop, so an answer that never comes freezes the
+/// whole TUI — no redraw, no keys. Generous enough for the slow ones (the
+/// `review.*` commands shell out to `git` and `gh`) and finite, which is the
+/// point.
+const REQUEST_TIMEOUT_SECS: u64 = 20;
+
+fn request_timeout() -> std::time::Duration {
+    let secs = std::env::var("BENTO_REQUEST_TIMEOUT_SECS")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(REQUEST_TIMEOUT_SECS);
+    std::time::Duration::from_secs(secs)
+}
+
 /// Send one request and return the `data` field of the response.
 pub(crate) async fn request_data(body: Value) -> std::io::Result<Value> {
+    tokio::time::timeout(request_timeout(), request_data_inner(body))
+        .await
+        .unwrap_or_else(|_| {
+            Err(std::io::Error::new(
+                std::io::ErrorKind::TimedOut,
+                "el daemon no respondió a tiempo",
+            ))
+        })
+}
+
+async fn request_data_inner(body: Value) -> std::io::Result<Value> {
     let mut stream = TcpStream::connect(addr()).await?;
     stream.write_all(body.to_string().as_bytes()).await?;
     stream.write_all(b"\n").await?;
@@ -212,3 +238,31 @@ pub(crate) fn print_help() {
     eprintln!();
     eprintln!("env: BENTO_DAEMON_ADDR (default 127.0.0.1:7877)");
 }
+
+#[cfg(test)]
+mod request_tests {
+    use super::*;
+
+    /// A daemon that accepts the connection and then never answers — exactly
+    /// what a hung `gh` call behind `review.prs` looks like from here.
+    #[tokio::test]
+    async fn a_daemon_that_never_answers_times_out_instead_of_hanging_forever() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.unwrap();
+            // Held open, silent, for longer than the timeout under test.
+            tokio::time::sleep(std::time::Duration::from_secs(30)).await;
+            drop(socket);
+        });
+        std::env::set_var("BENTO_DAEMON_ADDR", addr.to_string());
+        std::env::set_var("BENTO_REQUEST_TIMEOUT_SECS", "1");
+
+        let started = std::time::Instant::now();
+        let result = request_data(serde_json::json!({ "id": "1", "cmd": "review.prs" })).await;
+
+        assert!(result.is_err(), "una espera infinita congela el panel entero");
+        assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::TimedOut);
+        assert!(started.elapsed() < std::time::Duration::from_secs(5), "tardó demasiado en rendirse");
+    }
+}

daemon/bento-cli/src/review_stream.rs

@@ -7,6 +7,7 @@
 
 use serde_json::Value;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use bento_review::stream::{parse_stream_line, StreamLine};
 use tokio::net::TcpStream;
 use tokio::sync::mpsc;
 
@@ -58,9 +59,20 @@ async fn run(body: Value, tx: mpsc::UnboundedSender<ReviewEvent>) {
         match value.get("event").and_then(Value::as_str) {
             Some("review.output") => {
                 if let Some(chunk) = value.get("data").and_then(Value::as_str) {
-                    match classify_review_chunk(chunk) {
-                        ReviewChunk::Stdout(text) => { let _ = tx.send(ReviewEvent::Content(text.to_string())); }
-                        ReviewChunk::Stderr(msg) => { let _ = tx.send(ReviewEvent::Progress(msg)); }
+                    // Parsed by the shared crate, so the CLI, the phone client
+                    // and the desktop app cannot drift on the wire format.
+                    match parse_stream_line(chunk) {
+                        StreamLine::Text(text) => { let _ = tx.send(ReviewEvent::Content(text)); }
+                        StreamLine::Batch { index, total, label } => {
+                            let _ = tx.send(ReviewEvent::Progress(format!("BATCH:{index}/{total}:{label}")));
+                        }
+                        StreamLine::Synthesis => { let _ = tx.send(ReviewEvent::Progress("SYNTHESIS".into())); }
+                        StreamLine::Session { agent, id } => {
+                            let _ = tx.send(ReviewEvent::Progress(format!("SESSION:{agent}:{id}")));
+                        }
+                        StreamLine::Tool(tool) => { let _ = tx.send(ReviewEvent::Progress(tool)); }
+                        StreamLine::Error(message) => { let _ = tx.send(ReviewEvent::Progress(format!("error: {message}"))); }
+                        StreamLine::Done => {}
                     }
                 }
             }
@@ -71,63 +83,4 @@ async fn run(body: Value, tx: mpsc::UnboundedSender<ReviewEvent>) {
     let _ = tx.send(ReviewEvent::Done);
 }
 
-#[derive(Debug, PartialEq)]
-enum ReviewChunk<'a> {
-    Stderr(String),
-    Stdout(&'a str),
-}
-
-/// Routes the protocol's own control sentinels (batch/synthesis progress,
-/// the session-id marker, error text) away from the actual review content —
-/// mirrors the filtering `review.js` already does for the web panel.
-fn classify_review_chunk(chunk: &str) -> ReviewChunk<'_> {
-    let is_batch_or_session_marker = (chunk.starts_with("[BATCH:") || chunk.starts_with("[SESSION:"))
-        && chunk.ends_with(']');
-    if is_batch_or_session_marker || chunk == "[SYNTHESIS]" {
-        return ReviewChunk::Stderr(chunk.trim_start_matches('[').trim_end_matches(']').to_string());
-    }
-    if let Some(msg) = chunk.strip_prefix("[ERROR] ") {
-        return ReviewChunk::Stderr(format!("error: {msg}"));
-    }
-    // Las herramientas son progreso, no informe: enseñan qué está mirando el
-    // agente sin ensuciar el texto de la review.
-    if let Some(tool) = chunk.strip_prefix("[TOOL] ") {
-        return ReviewChunk::Stderr(tool.to_string());
-    }
-    ReviewChunk::Stdout(chunk)
-}
-
-#[cfg(test)]
-mod review_chunk_tests {
-    use super::*;
 
-    #[test]
-    fn batch_marker_goes_to_stderr() {
-        assert_eq!(classify_review_chunk("[BATCH:1/2]"), ReviewChunk::Stderr("BATCH:1/2".into()));
-    }
-
-    #[test]
-    fn session_marker_goes_to_stderr() {
-        assert_eq!(classify_review_chunk("[SESSION:claude:abc]"), ReviewChunk::Stderr("SESSION:claude:abc".into()));
-    }
-
-    #[test]
-    fn synthesis_marker_goes_to_stderr() {
-        assert_eq!(classify_review_chunk("[SYNTHESIS]"), ReviewChunk::Stderr("SYNTHESIS".into()));
-    }
-
-    #[test]
-    fn error_marker_is_prefixed_and_goes_to_stderr() {
-        assert_eq!(classify_review_chunk("[ERROR] algo falló"), ReviewChunk::Stderr("error: algo falló".into()));
-    }
-
-    #[test]
-    fn plain_text_goes_to_stdout() {
-        assert_eq!(classify_review_chunk("## Título"), ReviewChunk::Stdout("## Título"));
-    }
-
-    #[test]
-    fn bracketed_text_that_is_not_a_known_marker_goes_to_stdout() {
-        assert_eq!(classify_review_chunk("[foo]"), ReviewChunk::Stdout("[foo]"));
-    }
-}