stash

behavioral-specification.md

@@ -0,0 +1,189 @@
+# Yoke Behavioral Specification
+
+This document defines what yoke promises to its users. Every statement here is
+a testable invariant over observables — files, exit codes, process behavior.
+No statement references internal functions, line numbers, or implementation
+details. These invariants survive refactors and rewrites.
+
+---
+
+## 1. The Loop
+
+**Story:** You write a plan, a protocol, and some guards. You run `yoke run`.
+An agent executes your plan iteratively. Each iteration, it reads the protocol,
+does work, and updates notes. Guards check the work. When the agent writes
+`STATUS: DONE` and all guards pass, yoke exits.
+
+### Invariants
+
+**1.1 — The spec is immutable from the agent's perspective.**
+`protocol.md`, `plan.md`, and `yoke.conf` are backed up before the loop and
+restored before every iteration. The agent can overwrite them during its turn,
+but those changes do not persist to the next iteration.
+
+**1.2 — Termination requires both signals.**
+The loop only exits when `STATUS: DONE` appears in `notes.md` AND all guards
+pass. Neither condition alone is sufficient. If guards fail but status is DONE,
+the loop continues with feedback. If guards pass but status is not DONE, the
+loop continues.
+
+**1.3 — Guard feedback is visible.**
+`guard-results.md` is written after every iteration. The agent sees it on its
+next turn. No guard result is silently swallowed.
+
+**1.4 — Boundary violations block guards.**
+If the diff boundary check fails, all configured guards are skipped (not run).
+The agent gets boundary feedback only. Guards do not run on invalid state.
+
+**1.5 — Interrupts are clean.**
+SIGINT kills the child process immediately. The loop does not exit
+mid-iteration leaving partial state — it completes the signal check and exits
+at the next safe point with code 130.
+
+---
+
+## 2. The Judge
+
+**Story:** In brute mode, after the worker says DONE and guards pass, a
+separate fresh agent (the judge) runs. It reads `judge.md`, tests the feature,
+and writes `VERDICT: PASS` or `VERDICT: FAIL` to `verdict.md`. On PASS, yoke
+exits successfully. On FAIL, the worker retries.
+
+### Invariants
+
+**2.1 — The judge is independent.**
+It is a fresh agent invocation with no shared context from the worker. Its only
+input is `judge.md` and the codebase state.
+
+**2.2 — Verdict survives retries.**
+On judge FAIL, `verdict.md` is NOT cleared. The worker sees the judge's
+feedback on its next iteration. This is how the worker knows what went wrong.
+
+**2.3 — Guard results survive retries.**
+Same as verdict — `guard-results.md` persists across brute retries so the
+worker sees what the guards reported.
+
+**2.4 — Notes status reset on retry, nothing else.**
+On judge FAIL, only the first line of `notes.md` is overwritten to
+`STATUS: IN_PROGRESS`. The rest of the file — the agent's prior iteration
+notes — is preserved. All other files remain as-is. The worker starts with a
+clean status but full context from both its own notes and the judge's verdict.
+
+**2.5 — Bailout is exact.**
+If `max-judge-failures` consecutive judge FAILs occur, yoke exits non-zero.
+The count is exact — `max-judge-failures 2` means bailout on the 2nd
+consecutive FAIL, not the 3rd.
+
+**2.6 — Judge-every overrides cadence on DONE.**
+If `judge-every` is configured and the worker signals DONE, the judge fires
+immediately regardless of whether the iteration is on the cadence boundary.
+DONE always triggers judgment.
+
+---
+
+## 3. The Stash
+
+**Story:** `yoke stash` saves the current `.loop/` state. `yoke stash pop`
+restores the most recent snapshot. `yoke stash checkout <hash>` restores a
+specific snapshot. `yoke clean` auto-stashes before wiping.
+
+### Invariants
+
+**3.1 — Stash is a lossless round-trip.**
+`stash` then `pop` produces identical `.loop/` contents. No file is lost,
+truncated, or corrupted.
+
+**3.2 — Auto-stash before destructive operations.**
+Both `clean` and `checkout` auto-stash current state before modifying it. You
+can always recover what was there before.
+
+**3.3 — Mode tag is recorded.**
+Each stash entry records the mode (loop/brute/saga) that was active when it
+was created. This tag is preserved in the index and survives restore
+operations.
+
+**3.4 — Index is append-only.**
+Stash never modifies or deletes existing index lines. New entries are appended.
+The index is a history, not a mutable pointer.
+
+**3.5 — Prefix matching is unambiguous.**
+`checkout abc` matches any entry starting with `abc`. If multiple entries
+match, yoke errors instead of guessing. No silent wrong restore.
+
+---
+
+## 4. The Saga
+
+**Story:** Saga mode has a scoper agent that reads `specification.md`,
+decomposes it into chunks, writes each chunk to `sub-plan.md`, and a brute
+loop implements and verifies each chunk. When all chunks are done, the scoper
+writes `STATUS: DONE` to `saga-notes.md`.
+
+### Invariants
+
+**4.1 — Saga completion checks saga-notes, not notes.**
+The saga loop checks `saga-notes.md` for DONE. `notes.md` is local to each
+brute chunk and is cleared between chunks. Checking `notes.md` would be
+checking the wrong file.
+
+**4.2 — Brute bailout triggers re-scoping, not abort.**
+If brute fails `max-judge-failures` times on a chunk, control returns to the
+scoper. The scoper can re-scope the same chunk differently. The saga does not
+abort on a single chunk failure.
+
+**4.3 — Sub-plan must be non-empty.**
+If the scoper produces an empty `sub-plan.md`, the saga aborts. This prevents
+a brute loop from running with no plan.
+
+**4.4 — Chunk state is isolated but logged.** `notes.md`, `verdict.md`, and
+`guard-results.md` are cleared between chunks. Each brute run starts fresh.
+Previous chunk state does not leak into the next chunk. Before clearing,
+the contents of `notes.md` are appended to `saga-log.md`.
+
+**4.5 — Saga log is append-only.** `saga-log.md` accumulates the worker's
+notes from every completed chunk. It is never cleared or truncated during a
+saga run. Each entry is labeled with its chunk number.
+
+---
+
+## 5. Config
+
+**Story:** `yoke.conf` defines the rules of the loop — what files are
+protected, what guards run, how the judge behaves. It is parsed once at
+startup and applied consistently throughout the run.
+
+### Invariants
+
+**5.1 — Valid configs parse.**
+Every legal combination of directives parses without error.
+
+**5.2 — Invalid configs fail loudly.**
+Unknown directives, malformed values, and missing required fields produce clear
+errors — not silent defaults.
+
+**5.3 — Scope rules resolve most-specific-wins.**
+If `allow src/` and `no-modify src/main.rs` are both configured, `src/main.rs`
+is protected and `src/other.rs` is allowed. Longer prefix wins.
+
+**5.4 — Guard-after requires its periodic.**
+A `guard-after` referencing a periodic that does not exist is a config error,
+not a silent no-op.
+
+---
+
+## 6. Mode Switching
+
+**Story:** You can switch between loop, brute, and saga without losing
+progress. Each mode's state is snapshotted when you leave it and restored
+when you return.
+
+### Invariants
+
+**6.1 — Mode switch stashes current state.**
+Switching from mode A to B stashes all of A's files via `yoke stash`. The
+stash entry is tagged with mode A. Current state is always recoverable.
+
+**6.2 — Mode switch always fresh-inits.**
+After stashing, the target mode is initialized with fresh template files.
+Previous sessions are not auto-restored. Use `yoke stash checkout` to
+restore a prior session.

src/config.rs

@@ -41,24 +41,126 @@ pub struct Config {
     pub periodics: Vec<Periodic>,
 }
 
+struct ConfigBuilder {
+    max_tail: usize,
+    log_dir: Option<String>,
+    image: Option<String>,
+    model: Option<String>,
+    scope_rules: Vec<ScopeRule>,
+    guards: Vec<String>,
+    judge_every: Option<u32>,
+    max_judge_failures: u32,
+    periodics: Vec<Periodic>,
+    pending_guard_afters: Vec<(String, String, usize)>,
+}
+
+fn cfg_err(path: &Path, line_num: usize, msg: &str) -> String {
+    format!("{}:{}: {}", path.display(), line_num, msg)
+}
+
+fn parse_positive_u32(value: &str, path: &Path, line_num: usize, label: &str) -> Result<u32, String> {
+    let n = value.parse::<u32>()
+        .map_err(|_| cfg_err(path, line_num, &format!("invalid {} value '{}'", label, value)))?;
+    if n == 0 {
+        return Err(cfg_err(path, line_num, &format!("{} must be > 0", label)));
+    }
+    Ok(n)
+}
+
+#[allow(clippy::string_slice)]
+fn parse_periodic(value: &str, path: &Path, line_num: usize) -> Result<Periodic, String> {
+    let trimmed = value.trim();
+    let split_pos = trimmed.rfind(char::is_whitespace)
+        .ok_or_else(|| cfg_err(path, line_num, "periodic requires '<path> <cadence>'"))?;
+    let ppath = trimmed[..split_pos].trim();
+    let cadence = parse_positive_u32(trimmed[split_pos..].trim(), path, line_num, "periodic cadence")?;
+    let name = Path::new(ppath)
+        .file_stem()
+        .and_then(|s| s.to_str())
+        .ok_or_else(|| cfg_err(path, line_num, &format!("cannot derive name from periodic path '{}'", ppath)))?
+        .to_string();
+    Ok(Periodic { path: ppath.to_string(), name, cadence, guards: Vec::new() })
+}
+
+#[allow(clippy::string_slice)]
+fn parse_guard_after(value: &str, path: &Path, line_num: usize) -> Result<(String, String, usize), String> {
+    let trimmed = value.trim();
+    let split_pos = trimmed.find(char::is_whitespace)
+        .ok_or_else(|| cfg_err(path, line_num, "guard-after requires '<periodic-name> <command>'"))?;
+    let pname = trimmed[..split_pos].trim().to_string();
+    let cmd = trimmed[split_pos..].trim().to_string();
+    Ok((pname, cmd, line_num))
+}
+
+impl ConfigBuilder {
+    fn new() -> Self {
+        Self {
+            max_tail: 200,
+            log_dir: None,
+            image: None,
+            model: None,
+            scope_rules: Vec::new(),
+            guards: Vec::new(),
+            judge_every: None,
+            max_judge_failures: 3,
+            periodics: Vec::new(),
+            pending_guard_afters: Vec::new(),
+        }
+    }
+
+    #[allow(clippy::string_slice)]
+    fn parse_line(&mut self, directive: &str, value: &str, path: &Path, line_num: usize) -> Result<(), String> {
+        match directive {
+            "max-tail" => {
+                self.max_tail = value.parse::<usize>()
+                    .map_err(|_| cfg_err(path, line_num, &format!("invalid max-tail value '{}'", value)))?;
+            }
+            "log-dir" => self.log_dir = Some(value.to_string()),
+            "image" => self.image = Some(value.to_string()),
+            "model" => self.model = Some(value.to_string()),
+            "allow" => self.scope_rules.push(ScopeRule { tag: ScopeTag::Allow, prefix: value.to_string() }),
+            "add-only" => self.scope_rules.push(ScopeRule { tag: ScopeTag::AddOnly, prefix: value.to_string() }),
+            "no-modify" => self.scope_rules.push(ScopeRule { tag: ScopeTag::NoModify, prefix: value.to_string() }),
+            "guard" => self.guards.push(value.to_string()),
+            "judge-every" => self.judge_every = Some(parse_positive_u32(value, path, line_num, "judge-every")?),
+            "max-judge-failures" => self.max_judge_failures = parse_positive_u32(value, path, line_num, "max-judge-failures")?,
+            "periodic" => self.periodics.push(parse_periodic(value, path, line_num)?),
+            "guard-after" => self.pending_guard_afters.push(parse_guard_after(value, path, line_num)?),
+            other => return Err(cfg_err(path, line_num, &format!("unknown directive '{}'", other))),
+        }
+        Ok(())
+    }
+
+    fn build(self, path: &Path) -> Result<Config, String> {
+        let mut periodics = self.periodics;
+        for (pname, cmd, ln) in self.pending_guard_afters {
+            match periodics.iter_mut().find(|p| p.name == pname) {
+                Some(p) => p.guards.push(cmd),
+                None => return Err(cfg_err(path, ln, &format!("guard-after references unknown periodic '{}'", pname))),
+            }
+        }
+        Ok(Config {
+            max_tail: self.max_tail,
+            log_dir: self.log_dir,
+            image: self.image,
+            model: self.model,
+            scope_rules: self.scope_rules,
+            guards: self.guards,
+            judge_every: self.judge_every,
+            max_judge_failures: self.max_judge_failures,
+            periodics,
+        })
+    }
+}
+
 impl Config {
     #[allow(clippy::string_slice)] // all slices at ASCII delimiter positions from .find()
     pub fn load(path: &Path) -> Result<Config, String> {
         let content = fs::read_to_string(path)
             .map_err(|e| format!("failed to read config {}: {}", path.display(), e))?;
 
-        let mut max_tail: usize = 200;
-        let mut log_dir: Option<String> = None;
-        let mut image: Option<String> = None;
-        let mut model: Option<String> = None;
-        let mut scope_rules = Vec::new();
-        let mut guards = Vec::new();
-        let mut judge_every: Option<u32> = None;
-        let mut max_judge_failures: u32 = 3;
-        let mut periodics = Vec::new();
-        let mut pending_guard_afters: Vec<(String, String, usize)> = Vec::new(); // (periodic_name, command, line_num)
+        let mut builder = ConfigBuilder::new();
         for (line_num, raw_line) in content.lines().enumerate() {
-            // Strip comments
             let line = match raw_line.find('#') {
                 Some(pos) => &raw_line[..pos],
                 None => raw_line,
@@ -68,187 +170,17 @@ impl Config {
                 continue;
             }
 
-            // Split into directive and value at first whitespace
             let (directive, value) = match line.find(char::is_whitespace) {
                 Some(pos) => (&line[..pos], line[pos..].trim_start()),
                 None => {
-                    return Err(format!(
-                        "{}:{}: directive '{}' has no value",
-                        path.display(),
-                        line_num + 1,
-                        line
-                    ));
+                    return Err(cfg_err(path, line_num + 1, &format!("directive '{}' has no value", line)));
                 }
             };
 
-            match directive {
-                "max-tail" => {
-                    max_tail = value.parse::<usize>().map_err(|_| {
-                        format!(
-                            "{}:{}: invalid max-tail value '{}'",
-                            path.display(),
-                            line_num + 1,
-                            value
-                        )
-                    })?;
-                }
-                "log-dir" => {
-                    log_dir = Some(value.to_string());
-                }
-                "image" => {
-                    image = Some(value.to_string());
-                }
-                "model" => {
-                    model = Some(value.to_string());
-                }
-                "allow" => scope_rules.push(ScopeRule {
-                    tag: ScopeTag::Allow,
-                    prefix: value.to_string(),
-                }),
-                "add-only" => scope_rules.push(ScopeRule {
-                    tag: ScopeTag::AddOnly,
-                    prefix: value.to_string(),
-                }),
-                "no-modify" => scope_rules.push(ScopeRule {
-                    tag: ScopeTag::NoModify,
-                    prefix: value.to_string(),
-                }),
-                "guard" => guards.push(value.to_string()),
-                "judge-every" => {
-                    let n = value.parse::<u32>().map_err(|_| {
-                        format!(
-                            "{}:{}: invalid judge-every value '{}'",
-                            path.display(),
-                            line_num + 1,
-                            value
-                        )
-                    })?;
-                    if n == 0 {
-                        return Err(format!(
-                            "{}:{}: judge-every must be > 0",
-                            path.display(),
-                            line_num + 1
-                        ));
-                    }
-                    judge_every = Some(n);
-                }
-                "max-judge-failures" => {
-                    let n = value.parse::<u32>().map_err(|_| {
-                        format!(
-                            "{}:{}: invalid max-judge-failures value '{}'",
-                            path.display(),
-                            line_num + 1,
-                            value
-                        )
-                    })?;
-                    if n == 0 {
-                        return Err(format!(
-                            "{}:{}: max-judge-failures must be > 0",
-                            path.display(),
-                            line_num + 1
-                        ));
-                    }
-                    max_judge_failures = n;
-                }
-                "periodic" => {
-                    // Split value at last whitespace → path + cadence
-                    let trimmed = value.trim();
-                    let split_pos = trimmed.rfind(char::is_whitespace).ok_or_else(|| {
-                        format!(
-                            "{}:{}: periodic requires '<path> <cadence>'",
-                            path.display(),
-                            line_num + 1
-                        )
-                    })?;
-                    let ppath = trimmed[..split_pos].trim();
-                    let cadence_str = trimmed[split_pos..].trim();
-                    let cadence = cadence_str.parse::<u32>().map_err(|_| {
-                        format!(
-                            "{}:{}: invalid periodic cadence '{}'",
-                            path.display(),
-                            line_num + 1,
-                            cadence_str
-                        )
-                    })?;
-                    if cadence == 0 {
-                        return Err(format!(
-                            "{}:{}: periodic cadence must be > 0",
-                            path.display(),
-                            line_num + 1
-                        ));
-                    }
-                    // Derive name from filename stem
-                    let name = Path::new(ppath)
-                        .file_stem()
-                        .and_then(|s| s.to_str())
-                        .ok_or_else(|| {
-                            format!(
-                                "{}:{}: cannot derive name from periodic path '{}'",
-                                path.display(),
-                                line_num + 1,
-                                ppath
-                            )
-                        })?
-                        .to_string();
-                    periodics.push(Periodic {
-                        path: ppath.to_string(),
-                        name,
-                        cadence,
-                        guards: Vec::new(),
-                    });
-                }
-                "guard-after" => {
-                    // Split at first whitespace → periodic name + command
-                    let trimmed = value.trim();
-                    let split_pos = trimmed.find(char::is_whitespace).ok_or_else(|| {
-                        format!(
-                            "{}:{}: guard-after requires '<periodic-name> <command>'",
-                            path.display(),
-                            line_num + 1
-                        )
-                    })?;
-                    let pname = trimmed[..split_pos].trim().to_string();
-                    let cmd = trimmed[split_pos..].trim().to_string();
-                    pending_guard_afters.push((pname, cmd, line_num + 1));
-                }
-                other => {
-                    return Err(format!(
-                        "{}:{}: unknown directive '{}'",
-                        path.display(),
-                        line_num + 1,
-                        other
-                    ));
-                }
-            }
+            builder.parse_line(directive, value, path, line_num + 1)?;
         }
 
-        // Attach guard-after commands to their matching periodics
-        for (pname, cmd, ln) in pending_guard_afters {
-            let found = periodics.iter_mut().find(|p| p.name == pname);
-            match found {
-                Some(p) => p.guards.push(cmd),
-                None => {
-                    return Err(format!(
-                        "{}:{}: guard-after references unknown periodic '{}'",
-                        path.display(),
-                        ln,
-                        pname
-                    ));
-                }
-            }
-        }
-
-        Ok(Config {
-            max_tail,
-            log_dir,
-            image,
-            model,
-            scope_rules,
-            guards,
-            judge_every,
-            max_judge_failures,
-            periodics,
-        })
+        builder.build(path)
     }
 
     /// Resolve the most-specific scope tag for a file path.

src/stash.rs

@@ -213,6 +213,56 @@ pub(crate) fn stash_log_cmd() -> i32 {
     0
 }
 
+/// Resolve a hash prefix to a single stash entry. Returns the entry or logs an error.
+fn resolve_stash_entry<'a>(entries: &'a [StashEntry], target_hash: &str) -> Option<&'a StashEntry> {
+    let matches: Vec<&StashEntry> = entries
+        .iter()
+        .filter(|e| e.hash == target_hash || e.hash.starts_with(target_hash))
+        .collect();
+
+    match matches.len() {
+        0 => {
+            log_error(&format!("no stash entry matching '{}'", target_hash));
+            None
+        }
+        1 => Some(matches[0]),
+        n => {
+            log_error(&format!(
+                "ambiguous hash '{}' — matches {} entries",
+                target_hash, n
+            ));
+            None
+        }
+    }
+}
+
+/// Auto-stash current state, then replace `.loop/` files with those from `entry_dir`.
+fn swap_loop_files(entry_dir: &Path, mode: &str) -> Result<(), String> {
+    let current_files = collect_stashable_files();
+    if !current_files.is_empty() {
+        let hash = stash_snapshot(mode)?;
+        log(&format!(
+            "auto-stashed current state → {}{}{}",
+            BLUE, hash, RESET
+        ));
+    }
+
+    for (name, _) in &collect_stashable_files() {
+        let _ = fs::remove_file(Path::new(".loop").join(name));
+    }
+
+    if let Ok(dir_entries) = fs::read_dir(entry_dir) {
+        for entry in dir_entries.flatten() {
+            let name = entry.file_name();
+            let dest = Path::new(".loop").join(&name);
+            fs::copy(entry.path(), &dest).map_err(|e| {
+                format!("failed to restore {}: {}", name.to_string_lossy(), e)
+            })?;
+        }
+    }
+    Ok(())
+}
+
 /// `yoke stash checkout <hash>` — auto-stash current state, clear `.loop/` files,
 /// restore target entry. Supports prefix matching.
 pub(crate) fn stash_checkout(target_hash: &str, mode: &str) -> i32 {
@@ -226,75 +276,24 @@ pub(crate) fn stash_checkout(target_hash: &str, mode: &str) -> i32 {
         return 1;
     }
 
-    // Find matching entries (exact or prefix)
-    let matches: Vec<&StashEntry> = entries
-        .iter()
-        .filter(|e| e.hash == target_hash || e.hash.starts_with(target_hash))
-        .collect();
+    let target = match resolve_stash_entry(&entries, target_hash) {
+        Some(t) => t,
+        None => return 1,
+    };
 
-    match matches.len() {
-        0 => {
-            log_error(&format!("no stash entry matching '{}'", target_hash));
-            1
-        }
-        1 => {
-            let target = matches[0];
-            let entry_dir = Path::new(STASH_DIR).join(&target.hash);
-            if !entry_dir.exists() {
-                log_error("stash entry directory missing — index is corrupt");
-                return 1;
-            }
-
-            // Auto-stash current state (skip if .loop/ has no stashable files)
-            let current_files = collect_stashable_files();
-            if !current_files.is_empty() {
-                match stash_snapshot(mode) {
-                    Ok(hash) => {
-                        log(&format!(
-                            "auto-stashed current state → {}{}{}",
-                            BLUE, hash, RESET
-                        ));
-                    }
-                    Err(msg) => {
-                        log_error(&format!("failed to auto-stash: {}", msg));
-                        return 1;
-                    }
-                }
-            }
-
-            // Remove all stashable files from .loop/
-            for (name, _) in &collect_stashable_files() {
-                let path = Path::new(".loop").join(name);
-                let _ = fs::remove_file(&path);
-            }
-
-            // Restore files from the target entry
-            if let Ok(dir_entries) = fs::read_dir(&entry_dir) {
-                for entry in dir_entries.flatten() {
-                    let name = entry.file_name();
-                    let dest = Path::new(".loop").join(&name);
-                    if let Err(e) = fs::copy(entry.path(), &dest) {
-                        log_error(&format!(
-                            "failed to restore {}: {}",
-                            name.to_string_lossy(),
-                            e
-                        ));
-                        return 1;
-                    }
-                }
-            }
-
-            log(&format!("checked out {}{}{}", BLUE, target.hash, RESET));
-            0
-        }
-        n => {
-            log_error(&format!(
-                "ambiguous hash '{}' — matches {} entries",
-                target_hash, n
-            ));
-            1
-        }
+    let entry_dir = Path::new(STASH_DIR).join(&target.hash);
+    if !entry_dir.exists() {
+        log_error("stash entry directory missing — index is corrupt");
+        return 1;
     }
+
+    if let Err(msg) = swap_loop_files(&entry_dir, mode) {
+        log_error(&msg);
+        return 1;
+    }
+
+    log(&format!("checked out {}{}{}", BLUE, target.hash, RESET));
+    0
 }
 
 /// `yoke stash pop` — checkout the most recent stash entry.

src/stream.rs

@@ -249,135 +249,113 @@ fn format_tool_call(line: &str) -> String {
     }
 }
 
+/// Handle "assistant" events: render tool_use summaries and track tool counts.
+fn handle_assistant(out: &mut (impl Write + ?Sized), line: &str, state: &mut StreamState) -> io::Result<()> {
+    if !line.contains("\"tool_use\"") {
+        return Ok(());
+    }
+    if let (Some(id), Some(name)) = (extract_str(line, "id"), extract_str(line, "name")) {
+        state.tool_use_names.insert(id.to_string(), name.to_string());
+        *state.tool_counts.entry(name.to_string()).or_insert(0) += 1;
+    }
+    let desc = format_tool_call(line);
+    writeln!(out, "  {}>>{} {}", GRAY, RESET, desc)
+}
+
+/// Handle "stream_event" events: render thinking timer, text deltas, and block boundaries.
+fn handle_stream_event(out: &mut (impl Write + ?Sized), line: &str, state: &mut StreamState) -> io::Result<()> {
+    if line.contains("\"content_block_delta\"") {
+        if line.contains("\"thinking_delta\"") {
+            if let Some(start) = state.thinking_start {
+                let elapsed = start.elapsed().as_secs_f64();
+                write!(out, "\r{}{}thinking {:.1}s{}", DIM, BLUE, elapsed, RESET)?;
+                out.flush()?;
+            }
+        } else if line.contains("\"text_delta\"")
+            && let Some(text) = extract_str(line, "text")
+        {
+            let text = unescape_json(text);
+            write!(out, "{}{}{}", DIM, text, RESET)?;
+            out.flush()?;
+        }
+    } else if line.contains("\"content_block_start\"") {
+        if line.contains("\"thinking\"") {
+            state.thinking_start = Some(Instant::now());
+            write!(out, "{}{}thinking 0.0s{}", DIM, BLUE, RESET)?;
+            out.flush()?;
+            state.in_thinking = true;
+        } else if !line.contains("\"tool_use\"") {
+            writeln!(out)?;
+        }
+    } else if line.contains("\"content_block_stop\"") {
+        if state.in_thinking {
+            if let Some(start) = state.thinking_start {
+                let elapsed = start.elapsed().as_secs_f64();
+                write!(out, "\r{}{}thinking {:.1}s{}", DIM, BLUE, elapsed, RESET)?;
+            }
+            state.in_thinking = false;
+            state.thinking_start = None;
+        }
+        writeln!(out)?;
+    }
+    Ok(())
+}
+
+/// Handle "user" events: render tool_result success/error badges.
+fn handle_tool_result(out: &mut (impl Write + ?Sized), line: &str, state: &StreamState) -> io::Result<()> {
+    if !line.contains("\"tool_result\"") {
+        return Ok(());
+    }
+    let is_error = line.contains("\"is_error\":true") || line.contains("\"is_error\": true");
+    if is_error {
+        writeln!(out, "  {}← {}{}✗{}", GRAY, RESET, RED, RESET)?;
+        let tail = format_bash_error_tail(line);
+        if !tail.is_empty() {
+            writeln!(out, "{}", tail)?;
+        }
+        return Ok(());
+    }
+    let tool_name = extract_str(line, "tool_use_id")
+        .and_then(|id| state.tool_use_names.get(id))
+        .map(|s| s.as_str());
+    let badge = match tool_name {
+        Some(name @ ("Grep" | "Glob")) => format_grep_glob_badge(name, line),
+        _ => String::new(),
+    };
+    if badge.is_empty() {
+        writeln!(out, "  {}← {}✓{}", GRAY, GREEN, RESET)
+    } else {
+        writeln!(out, "  {}← {}✓{} {}", GRAY, GREEN, RESET, badge)
+    }
+}
+
 /// Process a single NDJSON line, writing formatted output to `out`.
 /// `prior_total` is the accumulated cost from previous iterations, used to display a running total.
 /// Returns `Err` on write failure (e.g. broken pipe) so the caller can stop.
-fn process_line(out: &mut impl Write, line: &str, state: &mut StreamState, prior_total: f64) -> io::Result<()> {
+fn process_line(out: &mut (impl Write + ?Sized), line: &str, state: &mut StreamState, prior_total: f64) -> io::Result<()> {
     // Check for turn boundary (message_id change)
     if let Some(msg_id) = extract_str(line, "message_id") {
-        let changed = match &state.current_msg_id {
-            Some(prev) => prev != msg_id,
-            None => true,
-        };
-        if changed {
+        let is_new = state.current_msg_id.as_deref() != Some(msg_id);
+        if is_new {
             state.current_msg_id = Some(msg_id.to_string());
             state.turn_num += 1;
-            writeln!(
-                out,
-                "{}{}━━━ Turn {} ━━━{}",
-                BOLD, ORANGE, state.turn_num, RESET
-            )?;
+            writeln!(out, "{}{}━━━ Turn {} ━━━{}", BOLD, ORANGE, state.turn_num, RESET)?;
         }
     }
 
-    let ev_type = extract_str(line, "type");
-
-    match ev_type {
-        // system → check subtype for init
+    match extract_str(line, "type") {
         Some("system") => {
-            let ev_subtype = extract_str(line, "subtype");
-            if ev_subtype == Some("init") && !state.seen_init {
+            if extract_str(line, "subtype") == Some("init") && !state.seen_init {
                 state.seen_init = true;
                 let sid = extract_str(line, "session_id").unwrap_or("?");
                 let sid_short: String = sid.chars().take(12).collect();
-                let sid_short = sid_short.as_str();
                 let model = extract_str(line, "model").unwrap_or("?");
-                writeln!(
-                    out,
-                    "{}{}[stream]{} session {}… model={}",
-                    ORANGE, BOLD, RESET, sid_short, model
-                )?;
+                writeln!(out, "{}{}[stream]{} session {}… model={}", ORANGE, BOLD, RESET, sid_short, model)?;
             }
         }
-        // assistant → only tool_use summaries (text already shown via stream_event deltas)
-        Some("assistant") => {
-            if line.contains("\"tool_use\"") {
-                // Track tool_use id → name for badge display on tool_result
-                if let (Some(id), Some(name)) = (extract_str(line, "id"), extract_str(line, "name")) {
-                    state.tool_use_names.insert(id.to_string(), name.to_string());
-                    // Increment tool-use counter for iteration summary
-                    *state.tool_counts.entry(name.to_string()).or_insert(0) += 1;
-                }
-                let desc = format_tool_call(line);
-                writeln!(out, "  {}>>{} {}", GRAY, RESET, desc)?;
-            }
-        }
-        // stream_event → streaming deltas
-        Some("stream_event") => {
-            if line.contains("\"content_block_delta\"") {
-                if line.contains("\"thinking_delta\"") {
-                    // Show elapsed timer during extended thinking, rewriting in-place
-                    if let Some(start) = state.thinking_start {
-                        let elapsed = start.elapsed().as_secs_f64();
-                        write!(out, "\r{}{}thinking {:.1}s{}", DIM, BLUE, elapsed, RESET)?;
-                        out.flush()?;
-                    }
-                } else if line.contains("\"text_delta\"")
-                    && let Some(text) = extract_str(line, "text")
-                {
-                    let text = unescape_json(text);
-                    write!(out, "{}{}{}", DIM, text, RESET)?;
-                    out.flush()?;
-                }
-                // input_json_delta → skip silently
-            } else if line.contains("\"content_block_start\"") {
-                if line.contains("\"thinking\"") {
-                    state.thinking_start = Some(Instant::now());
-                    write!(out, "{}{}thinking 0.0s{}", DIM, BLUE, RESET)?;
-                    out.flush()?;
-                    state.in_thinking = true;
-                } else if !line.contains("\"tool_use\"") {
-                    writeln!(out)?;
-                }
-            } else if line.contains("\"content_block_stop\"") {
-                if state.in_thinking {
-                    // Print final elapsed time and end the line
-                    if let Some(start) = state.thinking_start {
-                        let elapsed = start.elapsed().as_secs_f64();
-                        write!(out, "\r{}{}thinking {:.1}s{}", DIM, BLUE, elapsed, RESET)?;
-                    }
-                    state.in_thinking = false;
-                    state.thinking_start = None;
-                }
-                writeln!(out)?;
-            }
-            // message_start, message_delta, message_stop → skip
-        }
-        // user → tool result summaries
-        Some("user") => {
-            if line.contains("\"tool_result\"") {
-                let is_error = line.contains("\"is_error\":true")
-                    || line.contains("\"is_error\": true");
-                if is_error {
-                    let tail = format_bash_error_tail(line);
-                    if tail.is_empty() {
-                        writeln!(out, "  {}← {}{}✗{}", GRAY, RESET, RED, RESET)?;
-                    } else {
-                        writeln!(out, "  {}← {}{}✗{}", GRAY, RESET, RED, RESET)?;
-                        writeln!(out, "{}", tail)?;
-                    }
-                } else {
-                    // Check if this is a Grep or Glob result for badge display
-                    let tool_name = extract_str(line, "tool_use_id")
-                        .and_then(|id| state.tool_use_names.get(id))
-                        .map(|s| s.as_str());
-                    match tool_name {
-                        Some("Grep") | Some("Glob") => {
-                            let badge = format_grep_glob_badge(tool_name.unwrap(), line);
-                            if badge.is_empty() {
-                                writeln!(out, "  {}← {}✓{}", GRAY, GREEN, RESET)?;
-                            } else {
-                                writeln!(out, "  {}← {}✓{} {}", GRAY, GREEN, RESET, badge)?;
-                            }
-                        }
-                        _ => {
-                            writeln!(out, "  {}← {}✓{}", GRAY, GREEN, RESET)?;
-                        }
-                    }
-                }
-            }
-        }
-        // result → green bold summary with running total
+        Some("assistant") => handle_assistant(out, line, state)?,
+        Some("stream_event") => handle_stream_event(out, line, state)?,
+        Some("user") => handle_tool_result(out, line, state)?,
         Some("result") => {
             let cost = extract_num(line, "cost_usd").unwrap_or(0.0);
             state.iteration_cost = cost;
@@ -392,12 +370,10 @@ fn process_line(out: &mut impl Write, line: &str, state: &mut StreamState, prior
                 ORANGE, BOLD, RESET, cost, total, turns, dur_secs
             )?;
         }
-        // Non-JSON or unrecognized — dim passthrough
-        _ => {
-            if ev_type.is_none() && !line.trim().is_empty() {
-                writeln!(out, "{}  {}{}", DIM, line.trim(), RESET)?;
-            }
+        None if !line.trim().is_empty() => {
+            writeln!(out, "{}  {}{}", DIM, line.trim(), RESET)?;
         }
+        _ => {}
     }
     Ok(())
 }
@@ -434,13 +410,20 @@ fn format_summary_strip(state: &StreamState) -> String {
     format!("{}  ⟪ {} ⟫{}", DIM, parts.join(" │ "), RESET)
 }
 
-/// Filter NDJSON stream from Claude and format as rich ANSI output on stdout.
-/// Consumes the stream entirely — raw NDJSON is not written to disk.
-/// `prior_total` is the accumulated cost from previous iterations.
-/// Returns the cost of this iteration (from the `result` event).
-pub fn filter_stream(stdout: ChildStdout, log_path: Option<&Path>, prior_total: f64) -> f64 {
+/// Shared stream loop: reads lines from stdout, tees to log, calls processor per line,
+/// prints a summary strip, and returns the iteration cost.
+///
+/// Used by both Claude and OpenCode stream filters to avoid duplicating the
+/// BufReader/signal-check/log-tee boilerplate.
+pub fn run_stream_loop<S>(
+    stdout: ChildStdout,
+    log_path: Option<&Path>,
+    state: &mut S,
+    mut process: impl FnMut(&mut dyn Write, &str, &mut S) -> io::Result<()>,
+    summarize: impl FnOnce(&S) -> Option<String>,
+    get_cost: impl FnOnce(&S) -> f64,
+) -> f64 {
     let reader = BufReader::new(stdout);
-    let mut state = StreamState::new();
     let mut log_file = log_path.and_then(|p| {
         std::fs::create_dir_all(p.parent().unwrap_or(Path::new("."))).ok();
         std::fs::File::create(p).ok()
@@ -462,22 +445,34 @@ pub fn filter_stream(stdout: ChildStdout, log_path: Option<&Path>, prior_total:
             continue;
         }
 
-        // Tee raw NDJSON to log file
         if let Some(ref mut f) = log_file {
             let _ = writeln!(f, "{}", line);
         }
 
-        if process_line(&mut out, &line, &mut state, prior_total).is_err() {
-            break; // stdout broken (e.g. pipe closed) — stop gracefully
+        if process(&mut out, &line, state).is_err() {
+            break;
         }
     }
 
-    // Print iteration summary strip after streaming ends (before guards)
-    if state.turn_num > 0 {
-        let strip = format_summary_strip(&state);
+    if let Some(strip) = summarize(state) {
         let _ = writeln!(out, "{}", strip);
     }
 
     let _ = out.flush();
-    state.iteration_cost
+    get_cost(state)
+}
+
+/// Filter NDJSON stream from Claude and format as rich ANSI output on stdout.
+/// `prior_total` is the accumulated cost from previous iterations.
+/// Returns the cost of this iteration (from the `result` event).
+pub fn filter_stream(stdout: ChildStdout, log_path: Option<&Path>, prior_total: f64) -> f64 {
+    let mut state = StreamState::new();
+    run_stream_loop(
+        stdout,
+        log_path,
+        &mut state,
+        |out, line, st| process_line(out, line, st, prior_total),
+        |st| if st.turn_num > 0 { Some(format_summary_strip(st)) } else { None },
+        |st| st.iteration_cost,
+    )
 }

src/stream_opencode.rs

@@ -1,5 +1,5 @@
 use std::collections::HashMap;
-use std::io::{self, BufRead, BufReader, Write};
+use std::io::{self, Write};
 use std::path::Path;
 use std::process::ChildStdout;
 
@@ -65,7 +65,7 @@ fn format_tool_call(tool_name: &str, input: &str) -> String {
     }
 }
 
-fn process_line(out: &mut impl Write, line: &str, state: &mut StreamState) -> io::Result<()> {
+fn process_line(out: &mut (impl Write + ?Sized), line: &str, state: &mut StreamState) -> io::Result<()> {
     let ev_type = extract_str(line, "type");
 
     match ev_type {
@@ -167,43 +167,13 @@ fn format_summary_strip(state: &StreamState) -> String {
 }
 
 pub fn filter_stream(stdout: ChildStdout, log_path: Option<&Path>, _prior_total: f64) -> f64 {
-    let reader = BufReader::new(stdout);
     let mut state = StreamState::new();
-    let mut log_file = log_path.and_then(|p| {
-        std::fs::create_dir_all(p.parent().unwrap_or(Path::new("."))).ok();
-        std::fs::File::create(p).ok()
-    });
-
-    let mut out = io::stdout().lock();
-
-    for line_result in reader.lines() {
-        if crate::signal::interrupted() {
-            break;
-        }
-
-        let line = match line_result {
-            Ok(l) => l,
-            Err(_) => break,
-        };
-
-        if line.trim().is_empty() {
-            continue;
-        }
-
-        if let Some(ref mut f) = log_file {
-            let _ = writeln!(f, "{}", line);
-        }
-
-        if process_line(&mut out, &line, &mut state).is_err() {
-            break;
-        }
-    }
-
-    if state.turn_num > 0 {
-        let strip = format_summary_strip(&state);
-        let _ = writeln!(out, "{}", strip);
-    }
-
-    let _ = out.flush();
-    state.iteration_cost
+    crate::stream::run_stream_loop(
+        stdout,
+        log_path,
+        &mut state,
+        |out, line, st| process_line(out, line, st),
+        |st| if st.turn_num > 0 { Some(format_summary_strip(st)) } else { None },
+        |st| st.iteration_cost,
+    )
 }