agentmux_srv/backend/reactive/

progress_watcher.rs

// Copyright 2025-2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! Pushed per-agent **progress**: the agent's current todo checklist and the
//! tool it is running right now, published as an `agent:progress` MuxEvent so
//! the Swarm pane can list them under the agent's name.
//!
//! Sibling of [`super::activity_watcher`], and deliberately a separate loop:
//! that one spends a Haiku call per agent to write an English one-liner and is
//! throttled accordingly. This one only reads NEW bytes of a block's `output`
//! file and parses JSON, so it costs nothing per tick and can run far more
//! often — which is the point, since a checklist that lags 20s behind the agent
//! is worse than no checklist.
//!
//! Why the backend and not the frontend: `agent-document-store` already holds
//! every tool call, but only for panes that are currently MOUNTED. The Swarm
//! pane's whole value is seeing agents you are not looking at, so the data has
//! to come from the same place the shell/cron/summary rows come from.
//!
//! **Why state is carried across ticks** (reagent P1 on PR #2952): the
//! checklist cannot be rebuilt from a fixed tail window. `TodoWrite`-style
//! calls are immune — one call carries the whole list — but the vocabulary
//! AgentMux actually ships against (`TaskCreate`/`TaskUpdate`) describes ONE
//! item per call, so reconstructing the list needs every still-open task's
//! original `TaskCreate` to be inside the window. A verbose `Bash` or a large
//! `Read` pushes those out within seconds, and the task would then vanish from
//! the checklist while still open, silently. So each block keeps an accumulated
//! [`BlockState`] and each tick only consumes the bytes appended since the last
//! one.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use tokio::time::interval;

use crate::backend::blockcontroller::{get_block_controller_status, STATUS_RUNNING};
use crate::backend::storage::filestore::FileStore;
use crate::backend::mps::Broker;
use crate::backend::work_facts::overlap_notes::{NewEdit, NewPr};

use super::get_global_handler;

/// How often to sweep registered agents. Much tighter than the summary loop's
/// 20s: this is an incremental read + JSON parse, with no model call to bill.
const SWEEP_INTERVAL_SECS: u64 = 3;

/// Most bytes consumed in one tick, and the most read when first catching up on
/// a block. A block seen from its first byte is exact; one that already had
/// more than this when we first saw it (srv restarted mid-session) is seeded
/// from its tail and flagged — see [`BlockState::partial`].
const MAX_READ_BYTES: i64 = 4 * 1024 * 1024;

/// Cap on rows published per agent, so one pathological checklist can't flood
/// the Swarm tree. Truncation is reported rather than silent — see
/// [`AgentProgress::todos_truncated`].
const MAX_TODOS: usize = 24;

/// Bound on remembered in-flight tool calls. Real concurrency is a handful; a
/// larger number means results are being missed, and the oldest entries are the
/// ones least likely to still be running.
const MAX_OPEN_TOOLS: usize = 32;

pub const EVENT_AGENT_PROGRESS: &str = "agent:progress";

/// Republish an unchanged payload every N ticks (~15s at a 3s sweep).
///
/// The change-suppression below is what keeps this loop quiet, but on its own
/// it strands a LATE subscriber (codex P1 on PR #2952): open the Swarm after an
/// agent's checklist has settled and there is no further change to deliver, so
/// the pane shows nothing indefinitely for a perfectly active agent. The event
/// is also published with `persist: 1` so the broker replays the latest one to
/// a freshly-subscribed route; this periodic republish is the belt to that
/// suspenders, and covers any subscriber the replay path does not.
const REPUBLISH_EVERY_TICKS: u64 = 5;

/// Tool names whose **call arguments** carry checklist state.
///
/// Deliberately a list rather than the single `TodoWrite` everyone expects:
/// Claude Code 2.1.177 — the version AgentMux ships against today — advertises
/// `TaskCreate`/`TaskUpdate` and no `TodoWrite` at all, and other providers
/// differ again. Matching by a known set plus the `Todo` substring below means
/// a version bump that renames the tool degrades to "no rows" rather than to
/// wrong rows.
///
/// READ-style tools (`TaskList`, `TodoRead`) are deliberately absent. Their
/// task data lives in the tool RESULT, not the call's `input`, and this parser
/// only reads `input` — so listing them would advertise support that silently
/// contributes nothing (reagent P2 on PR #2952). Supporting them means parsing
/// tool_result payloads, which is its own change.
const TODO_TOOL_NAMES: &[&str] = &["TodoWrite", "TaskCreate", "TaskUpdate"];

fn is_todo_tool(name: &str) -> bool {
    TODO_TOOL_NAMES.contains(&name) || name.contains("Todo")
}

/// Tools whose input names a file the agent is changing. Their paths become
/// the agent's recent files, which `backend::work_facts` compares across
/// agents (docs/specs/SPEC_AGENT_OVERLAP_AWARENESS_2026_10_10.md §3.1).
const FILE_EDIT_TOOL_NAMES: &[&str] = &["Edit", "Write", "MultiEdit", "NotebookEdit"];

/// A tool that runs a shell command (Claude's `Bash`, a provider's `shell`,
/// AgentMux's own `Shell`), by name.
fn is_shell_tool(name: &str) -> bool {
    let lower = name.to_ascii_lowercase();
    lower.contains("bash") || lower.ends_with("shell")
}

/// Whether a shell command opens a PR: `gh-agent pr create` or `gh pr create`,
/// anywhere in it (`cd x && gh-agent pr create --title …`, or inside a quoted
/// shell call like Codex's `pwsh -Command 'gh-agent pr create …'`), the
/// program named by its path or with `.cmd`/`.exe` too.
#[cfg(test)]
fn is_pr_create(command: &str) -> bool {
    pr_create(command).is_some()
}

/// The PR being opened in a shell command (see [`is_pr_create`]), with the
/// `--head` and `--base` it names, if any. Flags may sit between the words
/// (`gh --repo o/r pr create`, `gh pr -R o/r create`); the command ends at a
/// shell separator.
fn pr_create(command: &str) -> Option<NewPrCommand> {
    let clean = |w: &str| w.trim_matches(|c| matches!(c, '"' | '\'' | '(' | ')' | ';')).to_string();
    let raw: Vec<&str> = command.split_whitespace().collect();
    for (i, first) in raw.iter().enumerate() {
        let word = clean(first);
        let program = word.rsplit(['/', '\\']).next().unwrap_or(&word);
        let program = program.trim_end_matches(".cmd").trim_end_matches(".exe");
        if !matches!(program, "gh" | "gh-agent") {
            continue;
        }
        let mut rest: Vec<String> = Vec::new();
        for w in &raw[i + 1..] {
            if matches!(*w, "&&" | "||" | "|" | ";") {
                break;
            }
            rest.push(clean(w));
            if w.ends_with(';') {
                break;
            }
        }
        // `pr create`, skipping flags (`-R`/`--repo` take a value).
        let (mut k, mut words) = (0, Vec::new());
        while k < rest.len() && words.len() < 2 {
            match rest[k].as_str() {
                "-R" | "--repo" => k += 2,
                w if w.starts_with('-') => k += 1,
                w => {
                    words.push(w.to_string());
                    k += 1;
                }
            }
        }
        if words != ["pr", "create"] {
            continue;
        }
        let mut pr = NewPrCommand::default();
        while k < rest.len() {
            let w = rest[k].as_str();
            let next = || rest.get(k + 1).cloned();
            match w {
                "-H" | "--head" => {
                    pr.head = next();
                    k += 2;
                    continue;
                }
                "-B" | "--base" => {
                    pr.base = next();
                    k += 2;
                    continue;
                }
                _ => {}
            }
            if let Some(v) = w.strip_prefix("--head=") {
                pr.head = Some(v.to_string());
            } else if let Some(v) = w.strip_prefix("--base=") {
                pr.base = Some(v.to_string());
            }
            k += 1;
        }
        return Some(pr);
    }
    None
}

/// The head and base a PR-create command names (`--head`, `--base`).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct NewPrCommand {
    head: Option<String>,
    base: Option<String>,
}

/// PR creations kept per sweep, at most.
const MAX_NEW_PRS: usize = 4;

/// Most recent files kept per block, distinct paths, newest first.
pub const MAX_RECENT_FILES: usize = 50;

/// One file an agent changed, as its tool call named it.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RecentFile {
    /// The path exactly as the tool input wrote it (usually absolute).
    pub path: String,
    pub tool: String,
    /// When the call was made: the transcript line's own `timestamp`, else
    /// the moment this watcher read it live. A line without a timestamp read
    /// as backlog (see [`apply_lines`]) is not recorded at all, so a block
    /// first seen mid-session never reads as "edited just now".
    pub ts_ms: u64,
}

/// A transcript line's own time, when it carries one (`timestamp`, RFC 3339
/// or Unix ms, as Claude's session lines do).
fn line_timestamp(v: &serde_json::Value) -> Option<u64> {
    let ts = v.get("timestamp")?;
    if let Some(n) = ts.as_u64() {
        return Some(n);
    }
    chrono::DateTime::parse_from_rfc3339(ts.as_str()?)
        .ok()
        .and_then(|dt| u64::try_from(dt.timestamp_millis()).ok())
}

/// What the watcher currently knows about one block, for readers outside
/// the sweep loop.
#[derive(Debug, Clone, Default)]
pub struct WatchedWork {
    pub progress: AgentProgress,
    pub recent_files: Vec<RecentFile>,
}

/// The latest [`WatchedWork`] per block id, replaced after every sweep.
fn snapshot() -> &'static parking_lot::RwLock<HashMap<String, WatchedWork>> {
    static SNAPSHOT: std::sync::LazyLock<parking_lot::RwLock<HashMap<String, WatchedWork>>> =
        std::sync::LazyLock::new(Default::default);
    &SNAPSHOT
}

/// Read access to the watcher's state for one block: its checklist, current
/// tool and recent files. `None` until the watcher has swept it once.
pub fn watched_work(block_id: &str) -> Option<WatchedWork> {
    snapshot().read().get(block_id).cloned()
}

/// Record that `path` was just changed by `tool`: moved to the front if it
/// was already listed, and the list kept to [`MAX_RECENT_FILES`].
fn note_recent_file(files: &mut Vec<RecentFile>, path: &str, tool: &str, ts_ms: u64) {
    files.retain(|f| f.path != path);
    files.insert(0, RecentFile { path: path.to_string(), tool: tool.to_string(), ts_ms });
    files.truncate(MAX_RECENT_FILES);
}

/// One checklist entry as the Swarm renders it.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, ts_rs::TS)]
#[ts(export, export_to = "../../../frontend/types/rpc/")]
pub struct TodoItem {
    pub text: String,
    /// `pending` | `in_progress` | `completed`. Passed through as the provider
    /// wrote it where possible; unknown values are kept verbatim rather than
    /// coerced, so a new provider status shows up as itself instead of
    /// silently becoming "pending".
    pub status: String,
}

/// What we currently believe about one agent.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
pub struct AgentProgress {
    pub todos: Vec<TodoItem>,
    /// How many rows [`MAX_TODOS`] dropped, so the UI can say "+N more"
    /// instead of quietly showing a partial list as if it were the whole one.
    pub todos_truncated: usize,
    /// True when this block was first seen mid-stream, so item-at-a-time
    /// (`TaskCreate`) calls from before that point were never observed and the
    /// checklist may be missing entries. Distinct from `todos_truncated`,
    /// which is a cap we chose; this is history we could not see.
    pub todos_partial: bool,
    /// The tool this agent is running right now — the oldest `tool_use` with no
    /// matching `tool_result` yet. `None` when the agent is between tools.
    pub current_tool: Option<String>,
}

impl AgentProgress {
    /// Nothing worth publishing — avoids a MuxEvent per tick per idle agent.
    fn is_empty(&self) -> bool {
        self.todos.is_empty() && self.current_tool.is_none()
    }
}

/// Everything we accumulate for one block across ticks.
#[derive(Debug, Default)]
struct BlockState {
    /// Byte offset of the next unread line in the block's `output`.
    next_offset: i64,
    /// Item-at-a-time accumulation (`TaskCreate`/`TaskUpdate`), keyed so an
    /// update revises its own row. Insertion order is the agent's own creation
    /// order, the only ordering it ever expressed.
    tasks: Vec<(String, TodoItem)>,
    /// Last whole-list call. Supersedes `tasks` outright when present, because
    /// that is the tool's own replace-everything semantics.
    list: Option<Vec<TodoItem>>,
    /// In-flight `tool_use` calls, oldest first.
    open: Vec<(String, String)>,
    /// See [`AgentProgress::todos_partial`].
    partial: bool,
    /// Files named by edit tools, newest first (see [`note_recent_file`]).
    recent_files: Vec<RecentFile>,
    /// Edits read live since the sweep last took them, oldest first: what
    /// the overlap notes check (`work_facts::overlap_notes`).
    new_edits: Vec<RecentFile>,
    /// PRs it started opening, read live since the sweep last took them: what
    /// the PR overlap note checks (`work_facts::overlap_notes::on_new_prs`).
    new_prs: Vec<NewPrCommand>,
    /// This block's output has been looked at before.
    seen: bool,
}

impl BlockState {
    fn progress(&self) -> AgentProgress {
        let mut todos = self
            .list
            .clone()
            .unwrap_or_else(|| self.tasks.iter().map(|(_, i)| i.clone()).collect());
        let todos_truncated = todos.len().saturating_sub(MAX_TODOS);
        todos.truncate(MAX_TODOS);
        AgentProgress {
            todos,
            todos_truncated,
            // A whole-list call is self-contained, so once one has been seen
            // the checklist is complete regardless of what we missed earlier.
            todos_partial: self.partial && self.list.is_none(),
            current_tool: self.open.first().map(|(_, name)| name.clone()),
        }
    }
}

/// Fold newly-appended transcript lines into `state`.
///
/// Pure w.r.t. I/O so it can be unit-tested against real transcript shapes
/// without a FileStore. Called once per tick with only the lines appended since
/// the previous call — never the whole file.
///
/// A recent file takes its line's own timestamp. Without one it takes
/// `live_now_ms`: the read time when these lines were just written, `None`
/// when they are backlog read on first sight of the block (srv restarted, or
/// the block predates the watcher), whose read time says nothing about when
/// the edit happened. Such an edit is not recorded.
fn apply_lines(state: &mut BlockState, lines: &[&str], live_now_ms: Option<u64>) {
    for line in lines {
        let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };

        // Resolve finished calls first: a tool_use and its result never share a
        // line, so ordering within the line is not a concern.
        for id in tool_result_ids(&v) {
            state.open.retain(|(open_id, _)| *open_id != id);
        }

        // Codex reports a shell command as an `item.started` event of type
        // `command_execution`, not a `tool_use`.
        if v.get("type").and_then(|t| t.as_str()) == Some("item.started") {
            let item = v.get("item");
            let is_command = item.and_then(|i| i.get("type")).and_then(|t| t.as_str()) == Some("command_execution");
            let command = item.and_then(|i| i.get("command")).and_then(|c| c.as_str());
            if let Some(pr) = command.filter(|_| is_command).and_then(pr_create) {
                if live_now_ms.is_some() && state.new_prs.len() < MAX_NEW_PRS {
                    state.new_prs.push(pr);
                }
            }
        }

        for frame in tool_frames(&v) {
            if frame.get("type").and_then(|t| t.as_str()) != Some("tool_use") {
                continue;
            }
            let Some(name) = frame_tool_name(frame) else { continue };
            if let Some(id) = frame_tool_id(frame) {
                state.open.push((id.to_string(), name.to_string()));
                // Drop the oldest if we're clearly missing results, rather than
                // letting this grow for the life of the block.
                if state.open.len() > MAX_OPEN_TOOLS {
                    state.open.remove(0);
                }
            }

            if is_shell_tool(name) {
                let opens = frame_input(frame).and_then(|i| first_string(i, &["command", "cmd"])).and_then(|c| pr_create(&c));
                if let Some(pr) = opens {
                    if live_now_ms.is_some() && state.new_prs.len() < MAX_NEW_PRS {
                        state.new_prs.push(pr);
                    }
                }
                continue;
            }

            if FILE_EDIT_TOOL_NAMES.contains(&name) {
                if let Some(path) = frame_input(frame)
                    .and_then(|i| first_string(i, &["file_path", "notebook_path"]))
                {
                    if let Some(ts) = line_timestamp(&v).or(live_now_ms) {
                        note_recent_file(&mut state.recent_files, &path, name, ts);
                        // Backlog is history, not something the agent is
                        // doing now: only live edits are checked for overlap.
                        if live_now_ms.is_some() && state.new_edits.len() < MAX_RECENT_FILES {
                            state.new_edits.push(RecentFile { path, tool: name.to_string(), ts_ms: ts });
                        }
                    }
                }
                continue;
            }

            if !is_todo_tool(name) {
                continue;
            }
            let Some(input) = frame_input(frame) else { continue };

            if let Some(items) = parse_todo_array(input) {
                state.list = Some(items);
                state.tasks.clear();
                continue;
            }
            // A whole-list call, once seen, owns the checklist — an
            // item-at-a-time call after it would be describing a different
            // vocabulary than the one currently in force.
            if state.list.is_none() {
                merge_single_task(&mut state.tasks, input, name);
            }
        }
    }
}

/// Every frame in a transcript line that could be a tool call or result.
///
/// Covers two genuinely different provider shapes (codex P2 on PR #2952):
///   * Claude stream-json — blocks inside `message.content` (or a bare
///     top-level `content` array).
///   * Gemini and friends — the LINE ITSELF is the frame:
///     `{"type":"tool_use","tool_name":…,"tool_id":…,"parameters":{…}}`
///     (see `frontend/app/view/agent/providers/gemini-translator.ts`).
///
/// Without the second, every non-Claude agent silently showed no in-flight
/// tool and no todos at all, since the array lookup just yielded nothing.
fn tool_frames(v: &serde_json::Value) -> Vec<&serde_json::Value> {
    let mut frames: Vec<&serde_json::Value> = v
        .get("message")
        .and_then(|m| m.get("content"))
        .or_else(|| v.get("content"))
        .and_then(|c| c.as_array())
        .map(|a| a.iter().collect())
        .unwrap_or_default();
    // A bare frame is only a frame if it declares one of the two types we
    // care about — otherwise every ordinary line would look like one.
    if matches!(v.get("type").and_then(|t| t.as_str()), Some("tool_use") | Some("tool_result")) {
        frames.push(v);
    }
    frames
}

/// Tool name under either vocabulary (`name` for Claude, `tool_name` for
/// Gemini-shaped frames).
fn frame_tool_name(frame: &serde_json::Value) -> Option<&str> {
    frame
        .get("name")
        .or_else(|| frame.get("tool_name"))
        .and_then(|n| n.as_str())
}

/// Call id under either vocabulary.
fn frame_tool_id(frame: &serde_json::Value) -> Option<&str> {
    frame
        .get("id")
        .or_else(|| frame.get("tool_id"))
        .and_then(|i| i.as_str())
}

/// Call arguments under either vocabulary.
fn frame_input(frame: &serde_json::Value) -> Option<&serde_json::Value> {
    frame.get("input").or_else(|| frame.get("parameters"))
}

fn tool_result_ids(v: &serde_json::Value) -> Vec<String> {
    let mut ids = Vec::new();
    if let Some(id) = v.get("tool_use_id").and_then(|i| i.as_str()) {
        ids.push(id.to_string());
    }
    for frame in tool_frames(v) {
        if frame.get("type").and_then(|t| t.as_str()) != Some("tool_result") {
            continue;
        }
        // `tool_use_id` (Claude) or `tool_id` (Gemini) — a result names the
        // call it belongs to under whichever key its provider uses.
        if let Some(id) = frame
            .get("tool_use_id")
            .or_else(|| frame.get("tool_id"))
            .and_then(|i| i.as_str())
        {
            ids.push(id.to_string());
        }
    }
    ids
}

/// `input.todos` (or `input.items`) as a whole checklist.
fn parse_todo_array(input: &serde_json::Value) -> Option<Vec<TodoItem>> {
    let arr = input
        .get("todos")
        .or_else(|| input.get("items"))
        .and_then(|t| t.as_array())?;
    // An empty array is a real state ("checklist cleared"), so return it as
    // such rather than falling through to the single-item path.
    Some(arr.iter().filter_map(parse_todo_entry).collect())
}

fn parse_todo_entry(entry: &serde_json::Value) -> Option<TodoItem> {
    let text = first_string(entry, &["content", "title", "text", "activeForm", "description"])?;
    let status = first_string(entry, &["status", "state"]).unwrap_or_else(|| "pending".to_string());
    Some(TodoItem { text, status })
}

/// Fold a Task-style call describing a SINGLE item into the accumulated rows.
///
/// Keyed by whatever id the tool uses so a later call revises the same row,
/// falling back to the text for tools that hand back no id.
///
/// Fields are merged, not replaced (codex P1 on PR #2952). A `TaskUpdate`
/// legitimately carries only `task_id` + the changed `status` — it has no
/// reason to resend text that did not change. Requiring a complete item meant
/// such a call parsed to nothing and the row sat at `pending` forever, which is
/// precisely the case a checklist exists to show moving.
/// Does this tool name describe CREATING an item, as opposed to revising one?
///
/// Only consulted for calls that carry no id (reagent P2 on PR #2952). With an
/// id the key is unambiguous; without one, text is the only identity available,
/// and keying on it alone silently collapses two genuinely distinct in-flight
/// tasks that happen to share a title. Two visible duplicate rows are a much
/// smaller harm than one invisibly missing row, so an idless CREATE always
/// appends and only an idless UPDATE matches by text.
fn is_create_call(tool_name: &str) -> bool {
    tool_name.contains("Create") || tool_name.contains("Add") || tool_name.contains("New")
}

fn merge_single_task(items: &mut Vec<(String, TodoItem)>, input: &serde_json::Value, tool_name: &str) {
    let text = first_string(input, &["content", "title", "text", "activeForm", "description"]);
    let status = first_string(input, &["status", "state"]);
    let id = first_string(input, &["task_id", "taskId", "id"]);

    // An idless create is its own row, always — see `is_create_call`.
    if id.is_none() && is_create_call(tool_name) {
        let Some(text) = text else { return };
        items.push((
            // Unique so no later call can match it by key; an idless tool
            // gives us nothing to correlate a future update against anyway.
            format!("{}#{}", text, items.len()),
            TodoItem { text, status: status.unwrap_or_else(|| "pending".to_string()) },
        ));
        return;
    }

    // Find the row this call refers to. An id matches the key it was stored
    // under; an idless update has only its text, and must compare against each
    // row's TEXT rather than its key — a row created WITH an id is keyed by
    // that id, so a key comparison would miss it and append a duplicate.
    let existing = match &id {
        Some(id) => items.iter_mut().find(|(k, _)| k == id),
        None => match &text {
            Some(text) => items.iter_mut().find(|(_, item)| item.text == *text),
            None => None,
        },
    };
    if let Some(slot) = existing {
        if let Some(text) = text {
            slot.1.text = text;
        }
        if let Some(status) = status {
            slot.1.status = status;
        }
        return;
    }

    let Some(key) = id.or_else(|| text.clone()) else {
        // Neither an id nor any text — nothing identifies a row to touch.
        return;
    };

    // Creating a row still needs something to display; an update for a task we
    // never saw created (seeded mid-stream) has no text to show.
    let Some(text) = text else { return };
    items.push((key, TodoItem { text, status: status.unwrap_or_else(|| "pending".to_string()) }));
}

fn first_string(v: &serde_json::Value, keys: &[&str]) -> Option<String> {
    keys.iter()
        .find_map(|k| v.get(*k).and_then(|s| s.as_str()))
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
}

/// Consume whatever has been appended to this block's `output` since the last
/// tick, folding it into `state`.
///
/// Returns false when there was nothing to do, so the caller can skip the rest
/// of the work for this block.
fn consume_new_output(filestore: &FileStore, block_id: &str, state: &mut BlockState) -> bool {
    let size = match filestore.stat(block_id, "output") {
        Ok(Some(wf)) => wf.size,
        Ok(None) => 0,
        Err(_) => return false,
    };
    // Whatever is already there the first time this block is looked at is
    // backlog; only what appears after that was written live (see
    // `apply_lines` on stamping recent files).
    let first_sight = !state.seen;
    state.seen = true;
    if size <= 0 {
        return false;
    }

    // The file shrank — a new session reusing the block, or a truncation. Our
    // accumulated state describes a transcript that no longer exists, so start
    // over rather than mixing two conversations' checklists.
    if size < state.next_offset {
        *state = BlockState { seen: true, ..BlockState::default() };
    }

    if state.next_offset == 0 && size > MAX_READ_BYTES {
        // First sight of a block that is already long (srv restarted
        // mid-session). Seed from the tail and admit the list may be missing
        // item-at-a-time entries from before this point.
        state.next_offset = size - MAX_READ_BYTES;
        state.partial = true;
    }

    let available = size - state.next_offset;
    if available <= 0 {
        return false;
    }
    let want = available.min(MAX_READ_BYTES);
    let Ok((_, bytes)) = filestore.read_at(block_id, "output", state.next_offset, want) else {
        return false;
    };
    if bytes.is_empty() {
        return false;
    }

    // Only consume up to the last complete line: a read boundary lands
    // anywhere, and half a JSON object would be dropped as malformed and then
    // never seen again in its complete form.
    let consumed = match bytes.iter().rposition(|b| *b == b'\n') {
        Some(idx) => idx + 1,
        // No newline anywhere in the chunk — an unterminated line longer than
        // the read. Skip it wholesale rather than stalling forever on it.
        None if want == MAX_READ_BYTES => bytes.len(),
        None => return false,
    };
    state.next_offset += consumed as i64;

    let text = String::from_utf8_lossy(&bytes[..consumed]);
    let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
    if lines.is_empty() {
        return false;
    }
    let live_now_ms = (!first_sight).then(agentmux_common::time::now_ms_u64);
    apply_lines(state, &lines, live_now_ms);
    true
}

/// Run the progress sweep loop. Never returns.
/// One block's worth of work decided by a sweep — what to publish, if anything.
struct SweepOutcome {
    agent_id: String,
    block_id: String,
    progress: AgentProgress,
}

/// One sweep's result: the states back, what to publish, and the edits and
/// PRs the overlap notes check.
type Swept = (HashMap<String, BlockState>, Vec<SweepOutcome>, (Vec<NewEdit>, Vec<NewPr>));

/// The whole file-reading half of a sweep, run OFF the async runtime.
///
/// `FileStore::stat`/`read_at` are synchronous, mutex-guarded SQLite reads, and
/// this loop runs them for every live agent at a 3s cadence (reagent P2 on PR
/// #2952). Doing that inline would park a Tokio worker thread on disk I/O
/// several times a second — tolerable at `activity_watcher`'s 20s, much less so
/// here. `states` is moved in and back out rather than shared behind a lock:
/// only this loop ever touches it, and one owner is simpler than one mutex.
fn sweep_blocking(
    filestore: &FileStore,
    agents: Vec<(String, String)>,
    mut states: HashMap<String, BlockState>,
    last_published: &HashMap<String, AgentProgress>,
    force_republish: bool,
) -> Swept {
    let mut out = Vec::new();
    let mut edits = Vec::new();
    let mut prs = Vec::new();

    for (agent_id, block_id) in agents {
        // Same gate as the summary loop: an idle or non-agent pane has no
        // progress to report and shouldn't be read every tick.
        let Some(status) = get_block_controller_status(&block_id) else { continue };
        if status.shellprocstatus != STATUS_RUNNING || !status.is_agent_pane {
            continue;
        }

        let state = states.entry(block_id.clone()).or_default();
        let had_new = consume_new_output(filestore, &block_id, state);
        edits.extend(state.new_edits.drain(..).map(|f| NewEdit {
            agent_id: agent_id.clone(),
            block_id: block_id.clone(),
            path: f.path,
        }));
        prs.extend(state.new_prs.drain(..).map(|p| NewPr {
            agent_id: agent_id.clone(),
            block_id: block_id.clone(),
            head: p.head,
            base: p.base,
        }));
        let progress = state.progress();

        // Nothing new AND nothing already published for this block — an agent
        // that has never had progress costs no events at all.
        if !had_new && !last_published.contains_key(&block_id) {
            continue;
        }
        if progress.is_empty() && !last_published.contains_key(&block_id) {
            continue;
        }
        if !force_republish && last_published.get(&block_id) == Some(&progress) {
            continue;
        }
        out.push(SweepOutcome { agent_id, block_id, progress });
    }

    (states, out, (edits, prs))
}

/// Run the progress sweep loop. Never returns.
pub async fn run_agent_progress_loop(filestore: Arc<FileStore>, broker: Arc<Broker>) {
    let mut ticker = interval(Duration::from_secs(SWEEP_INTERVAL_SECS));
    let mut tick: u64 = 0;
    let mut states: HashMap<String, BlockState> = HashMap::new();
    // Last payload published per block, so an unchanged checklist doesn't
    // re-publish every 3s.
    let mut last_published: HashMap<String, AgentProgress> = HashMap::new();

    loop {
        ticker.tick().await;
        tick += 1;
        let force_republish = tick % REPUBLISH_EVERY_TICKS == 0;

        let agents: Vec<(String, String)> = get_global_handler()
            .list_agents()
            .into_iter()
            .map(|a| (a.agent_id, a.block_id))
            .collect();
        let registered: std::collections::HashSet<String> =
            agents.iter().map(|(_, block_id)| block_id.clone()).collect();
        // Both maps are bounded by the live agent count, not by every block_id
        // ever seen in the process's lifetime.
        states.retain(|block_id, _| registered.contains(block_id));
        last_published.retain(|block_id, _| registered.contains(block_id));

        let fs = filestore.clone();
        let published = last_published.clone();
        let taken = std::mem::take(&mut states);
        let joined = tokio::task::spawn_blocking(move || {
            sweep_blocking(&fs, agents, taken, &published, force_republish)
        })
        .await;

        // A panic inside the blocking closure loses that tick's accumulated
        // state. Start clean rather than killing the loop: the next sweep
        // re-seeds from the tail (flagged partial), which is degraded but
        // recoverable, whereas returning here would end progress reporting for
        // the life of the process.
        let (returned_states, outcomes, (edits, prs)) = match joined {
            Ok(v) => v,
            Err(e) => {
                tracing::error!(error = %e, "progress sweep task failed; state reset for next tick");
                continue;
            }
        };
        states = returned_states;
        // Hands off and returns at once: the check runs on its own task.
        crate::backend::work_facts::overlap_notes::on_new_edits(edits);
        crate::backend::work_facts::overlap_notes::on_new_prs(prs);
        *snapshot().write() = states
            .iter()
            .map(|(block_id, s)| {
                (block_id.clone(), WatchedWork { progress: s.progress(), recent_files: s.recent_files.clone() })
            })
            .collect();

        for SweepOutcome { agent_id, block_id, progress } in outcomes {
            last_published.insert(block_id.clone(), progress.clone());

            let ts = agentmux_common::time::now_ms_u64();

            let event = crate::backend::mps_events::AgentProgressEvent {
                agent_id,
                block_id: block_id.clone(),
                todos: progress.todos,
                todos_truncated: progress.todos_truncated,
                todos_partial: progress.todos_partial,
                current_tool: progress.current_tool,
                ts,
            };
            // persist 1: replayed to a freshly-subscribed route, so a Swarm opened
            // mid-session sees the current checklist rather than waiting
            // for the agent's next change — see REPUBLISH_EVERY_TICKS.
            broker.publish_event::<crate::backend::mps_events::AgentProgress>(&event, vec![format!("block:{}", block_id)], 1);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn todo_write(id: &str, todos: &str) -> String {
        format!(
            r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"{id}","name":"TodoWrite","input":{{"todos":{todos}}}}}]}}}}"#
        )
    }

    fn task_call(id: &str, tool: &str, tid: &str, title: &str, status: &str) -> String {
        format!(
            r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"{id}","name":"{tool}","input":{{"task_id":"{tid}","title":"{title}","status":"{status}"}}}}]}}}}"#
        )
    }

    fn tool_use(id: &str, name: &str) -> String {
        format!(
            r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"{id}","name":"{name}","input":{{}}}}]}}}}"#
        )
    }

    fn tool_result(id: &str) -> String {
        format!(
            r#"{{"type":"user","message":{{"content":[{{"type":"tool_result","tool_use_id":"{id}"}}]}}}}"#
        )
    }

    /// One tick's worth of lines.
    fn feed(state: &mut BlockState, lines: &[String]) {
        feed_at(state, lines, 1_000);
    }

    fn feed_at(state: &mut BlockState, lines: &[String], now_ms: u64) {
        let refs: Vec<&str> = lines.iter().map(|s| s.as_str()).collect();
        apply_lines(state, &refs, Some(now_ms));
    }

    fn output_store(block: &str, text: &str) -> FileStore {
        use crate::backend::storage::filestore::{FileMeta, FileOpts};
        let fs = FileStore::open_in_memory().unwrap();
        fs.make_file(block, "output", FileMeta::default(), FileOpts::default()).unwrap();
        fs.append_data(block, "output", text.as_bytes()).unwrap();
        fs
    }

    /// An srv restart (or any first sight of a block) reads the backlog: an
    /// edit there without its own timestamp must not read as "edited now",
    /// one with a timestamp keeps it, and what is written afterwards is live.
    #[test]
    fn a_first_sight_backlog_does_not_read_as_just_edited() {
        let block = "b-backlog";
        let stamped = r#"{"type":"assistant","timestamp":"2026-10-01T10:00:00Z","message":{"content":[{"type":"tool_use","id":"s1","name":"Edit","input":{"file_path":"/r/stamped.rs"}}]}}"#;
        let backlog = format!("{}
{}
", edit_call("e1", "Edit", "file_path", "/r/old.rs"), stamped);
        let fs = output_store(block, &backlog);
        let mut st = BlockState::default();
        assert!(consume_new_output(&fs, block, &mut st));
        let got: Vec<(&str, u64)> = st.recent_files.iter().map(|f| (f.path.as_str(), f.ts_ms)).collect();
        assert_eq!(got, vec![("/r/stamped.rs", 1_790_848_800_000)], "the untimed backlog edit is not recorded");
        assert!(st.new_edits.is_empty(), "backlog edits are never checked for overlap");

        let before = agentmux_common::time::now_ms_u64();
        fs.append_data(block, "output", format!("{}
", edit_call("e2", "Write", "file_path", "/r/new.rs")).as_bytes())
            .unwrap();
        assert!(consume_new_output(&fs, block, &mut st));
        assert_eq!(st.recent_files[0].path, "/r/new.rs");
        assert!(st.recent_files[0].ts_ms >= before, "a live edit takes the read time");
        let live: Vec<&str> = st.new_edits.iter().map(|f| f.path.as_str()).collect();
        assert_eq!(live, vec!["/r/new.rs"], "a live edit is handed to the overlap check");
    }

    /// A block first seen while still empty: everything after is live.
    #[test]
    fn a_block_first_seen_empty_records_its_first_edits() {
        let block = "b-empty";
        let fs = FileStore::open_in_memory().unwrap();
        let mut st = BlockState::default();
        assert!(!consume_new_output(&fs, block, &mut st));
        let fs = output_store(block, &format!("{}
", edit_call("e1", "Edit", "file_path", "/r/a.rs")));
        assert!(consume_new_output(&fs, block, &mut st));
        assert_eq!(st.recent_files.len(), 1);
    }

    fn edit_call(id: &str, tool: &str, key: &str, path: &str) -> String {
        format!(
            r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"{id}","name":"{tool}","input":{{"{key}":"{path}","old_string":"a","new_string":"b"}}}}]}}}}"#
        )
    }

    /// A fake output stream with Edit / Write / MultiEdit / NotebookEdit
    /// calls fills recent files, newest first, with the read time.
    #[test]
    fn edit_tools_fill_recent_files_newest_first() {
        let mut st = BlockState::default();
        feed_at(
            &mut st,
            &[
                edit_call("e1", "Edit", "file_path", "C:/r/a.rs"),
                tool_result("e1"),
                edit_call("e2", "Write", "file_path", "C:/r/b.rs"),
                edit_call("e3", "MultiEdit", "file_path", "C:/r/c.rs"),
            ],
            10,
        );
        feed_at(&mut st, &[edit_call("e4", "NotebookEdit", "notebook_path", "C:/r/n.ipynb")], 20);
        let got: Vec<(&str, &str, u64)> =
            st.recent_files.iter().map(|f| (f.path.as_str(), f.tool.as_str(), f.ts_ms)).collect();
        assert_eq!(
            got,
            vec![
                ("C:/r/n.ipynb", "NotebookEdit", 20),
                ("C:/r/c.rs", "MultiEdit", 10),
                ("C:/r/b.rs", "Write", 10),
                ("C:/r/a.rs", "Edit", 10),
            ]
        );
        // The checklist and current tool are unchanged by the new bookkeeping.
        assert!(st.progress().todos.is_empty());
        assert_eq!(st.progress().current_tool.as_deref(), Some("Write"));
    }

    #[test]
    fn editing_a_file_again_moves_it_to_the_front_without_duplicating_it() {
        let mut st = BlockState::default();
        feed_at(&mut st, &[edit_call("e1", "Edit", "file_path", "/r/a.rs")], 1);
        feed_at(&mut st, &[edit_call("e2", "Edit", "file_path", "/r/b.rs")], 2);
        feed_at(&mut st, &[edit_call("e3", "Write", "file_path", "/r/a.rs")], 3);
        let got: Vec<(&str, u64)> = st.recent_files.iter().map(|f| (f.path.as_str(), f.ts_ms)).collect();
        assert_eq!(got, vec![("/r/a.rs", 3), ("/r/b.rs", 2)]);
        assert_eq!(st.recent_files[0].tool, "Write");
    }

    #[test]
    fn recent_files_stay_bounded() {
        let mut st = BlockState::default();
        let calls: Vec<String> = (0..MAX_RECENT_FILES + 7)
            .map(|i| edit_call(&format!("e{i}"), "Edit", "file_path", &format!("/r/f{i}.rs")))
            .collect();
        feed(&mut st, &calls);
        assert_eq!(st.recent_files.len(), MAX_RECENT_FILES);
        assert_eq!(st.recent_files[0].path, format!("/r/f{}.rs", MAX_RECENT_FILES + 6));
    }

    #[test]
    fn reads_and_other_tools_add_no_recent_files() {
        let mut st = BlockState::default();
        feed(
            &mut st,
            &[
                edit_call("r1", "Read", "file_path", "/r/a.rs"),
                edit_call("g1", "Grep", "file_path", "/r/b.rs"),
                edit_call("w1", "Edit", "old_string", "no path"),
            ],
        );
        assert!(st.recent_files.is_empty());
    }

    fn run(lines: &[String]) -> AgentProgress {
        let mut st = BlockState::default();
        feed(&mut st, lines);
        st.progress()
    }

    #[test]
    fn extracts_a_todowrite_checklist() {
        let p = run(&[todo_write(
            "t1",
            r#"[{"content":"Fix the bug","status":"in_progress"},{"content":"Write tests","status":"pending"}]"#,
        )]);
        assert_eq!(
            p.todos,
            vec![
                TodoItem { text: "Fix the bug".into(), status: "in_progress".into() },
                TodoItem { text: "Write tests".into(), status: "pending".into() },
            ]
        );
    }

    #[test]
    fn the_latest_full_list_supersedes_earlier_ones() {
        let p = run(&[
            todo_write("t1", r#"[{"content":"A","status":"pending"}, {"content":"B","status":"pending"}]"#),
            todo_write("t2", r#"[{"content":"A","status":"completed"}]"#),
        ]);
        assert_eq!(p.todos, vec![TodoItem { text: "A".into(), status: "completed".into() }]);
    }

    #[test]
    fn task_style_calls_accumulate_and_update_in_place() {
        let p = run(&[
            task_call("c1", "TaskCreate", "k1", "First", "pending"),
            task_call("c2", "TaskCreate", "k2", "Second", "pending"),
            task_call("u1", "TaskUpdate", "k1", "First", "completed"),
        ]);
        assert_eq!(
            p.todos,
            vec![
                TodoItem { text: "First".into(), status: "completed".into() },
                TodoItem { text: "Second".into(), status: "pending".into() },
            ],
            "an update revises its own row and preserves creation order",
        );
    }

    /// THE reagent P1 regression: a TaskCreate seen in an EARLIER tick must
    /// survive once its line has scrolled out of any fixed tail window. A
    /// stateless, tail-only parser drops it silently while the task is still
    /// open — the failure this whole accumulate-across-ticks design exists for.
    #[test]
    fn a_task_created_in_an_earlier_tick_survives_later_unrelated_output() {
        let mut st = BlockState::default();
        feed(&mut st, &[task_call("c1", "TaskCreate", "k1", "Long-lived task", "pending")]);

        // A later tick carrying only unrelated chatter — no todo tool in sight.
        let noise: Vec<String> = (0..50).map(|i| tool_use(&format!("n{i}"), "Read")).collect();
        feed(&mut st, &noise);

        assert_eq!(
            st.progress().todos,
            vec![TodoItem { text: "Long-lived task".into(), status: "pending".into() }],
            "the task is still open and must still be listed",
        );
    }

    #[test]
    fn an_update_arriving_a_tick_later_still_finds_its_row() {
        let mut st = BlockState::default();
        feed(&mut st, &[task_call("c1", "TaskCreate", "k1", "Task", "pending")]);
        feed(&mut st, &[tool_use("n1", "Bash"), tool_result("n1")]);
        feed(&mut st, &[task_call("u1", "TaskUpdate", "k1", "Task", "completed")]);

        assert_eq!(st.progress().todos, vec![TodoItem { text: "Task".into(), status: "completed".into() }]);
    }

    #[test]
    fn current_tool_is_the_outstanding_tool_use() {
        let p = run(&[tool_use("a", "Read"), tool_result("a"), tool_use("b", "Bash")]);
        assert_eq!(p.current_tool.as_deref(), Some("Bash"));
    }

    #[test]
    fn no_current_tool_once_every_call_has_returned() {
        let p = run(&[tool_use("a", "Read"), tool_result("a")]);
        assert_eq!(p.current_tool, None);
    }

    /// A result arriving in a later tick than its call must still clear it —
    /// the common case now that ticks are 3s and tools are slower than that.
    #[test]
    fn a_result_in_a_later_tick_clears_the_call_from_an_earlier_one() {
        let mut st = BlockState::default();
        feed(&mut st, &[tool_use("a", "Bash")]);
        assert_eq!(st.progress().current_tool.as_deref(), Some("Bash"));
        feed(&mut st, &[tool_result("a")]);
        assert_eq!(st.progress().current_tool, None);
    }

    #[test]
    fn an_explicitly_cleared_checklist_reads_as_empty() {
        let p = run(&[
            todo_write("t1", r#"[{"content":"A","status":"pending"}]"#),
            todo_write("t2", "[]"),
        ]);
        assert!(p.todos.is_empty());
    }

    #[test]
    fn truncates_a_pathological_list_and_reports_how_many_were_dropped() {
        let entries: Vec<String> = (0..MAX_TODOS + 5)
            .map(|i| format!(r#"{{"content":"item {i}","status":"pending"}}"#))
            .collect();
        let p = run(&[todo_write("t1", &format!("[{}]", entries.join(",")))]);
        assert_eq!(p.todos.len(), MAX_TODOS);
        assert_eq!(p.todos_truncated, 5);
    }

    /// `partial` is about history we could not see; `todos_truncated` is a cap
    /// we chose. They are different claims and must not be conflated.
    #[test]
    fn a_tail_seeded_block_reports_its_checklist_as_partial() {
        let mut st = BlockState { partial: true, ..Default::default() };
        feed(&mut st, &[task_call("c1", "TaskCreate", "k1", "Only what we saw", "pending")]);
        let p = st.progress();
        assert!(p.todos_partial, "item-at-a-time rows may be missing earlier entries");
        assert_eq!(p.todos_truncated, 0, "nothing was dropped by the cap");
    }

    /// …but a whole-list call is self-contained, so seeing one makes the
    /// checklist complete regardless of what came before.
    #[test]
    fn a_full_list_call_clears_the_partial_flag() {
        let mut st = BlockState { partial: true, ..Default::default() };
        feed(&mut st, &[todo_write("t1", r#"[{"content":"Whole list","status":"pending"}]"#)]);
        assert!(!st.progress().todos_partial);
    }

    #[test]
    fn ignores_malformed_lines_instead_of_giving_up_on_the_whole_batch() {
        let p = run(&[
            "not json at all".to_string(),
            String::new(),
            todo_write("t1", r#"[{"content":"Survives","status":"pending"}]"#),
        ]);
        assert_eq!(p.todos.len(), 1);
    }

    #[test]
    fn a_non_todo_tool_never_contributes_checklist_rows() {
        let p = run(&[tool_use("a", "Read"), tool_use("b", "Bash")]);
        assert!(p.todos.is_empty());
    }

    #[test]
    fn recognizes_todo_tools_by_name_across_provider_vocabularies() {
        assert!(is_todo_tool("TodoWrite"));
        assert!(is_todo_tool("TaskCreate"));
        assert!(is_todo_tool("TaskUpdate"));
        // Substring fallback for a renamed/namespaced variant.
        assert!(is_todo_tool("mcp__planner__TodoSync"));
        assert!(!is_todo_tool("Read"));
        assert!(!is_todo_tool("Task"), "the subagent-spawning Task tool is not a checklist");
    }

    /// Read-style tools return their data in the RESULT, which this parser
    /// never reads — so claiming to support them would be advertising a
    /// no-op (reagent P2 on PR #2952).
    #[test]
    fn read_style_task_tools_are_not_claimed_as_supported() {
        assert!(
            !TODO_TOOL_NAMES.contains(&"TaskList"),
            "TaskList's tasks live in its tool_result, not its input",
        );
        // TodoRead matches only via the deliberate `Todo` substring fallback,
        // and still contributes nothing because its input carries no items —
        // which is correct behavior, not a supported path.
        assert!(parse_todo_array(&serde_json::json!({})).is_none());
    }

    #[test]
    fn the_open_tool_list_stays_bounded_when_results_go_missing() {
        let mut st = BlockState::default();
        let calls: Vec<String> = (0..MAX_OPEN_TOOLS + 20)
            .map(|i| tool_use(&format!("t{i}"), "Read"))
            .collect();
        feed(&mut st, &calls);
        assert_eq!(st.open.len(), MAX_OPEN_TOOLS);
    }

    /// codex P1: a TaskUpdate legitimately carries only the id and the changed
    /// status. Requiring a complete item meant the row sat at `pending`
    /// forever — precisely the transition a checklist exists to show.
    #[test]
    fn a_status_only_update_moves_the_row_without_resending_its_text() {
        let mut st = BlockState::default();
        feed(&mut st, &[task_call("c1", "TaskCreate", "k1", "Ship it", "pending")]);
        feed(
            &mut st,
            &[format!(
                r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"u1","name":"TaskUpdate","input":{{"task_id":"k1","status":"completed"}}}}]}}}}"#
            )],
        );
        assert_eq!(
            st.progress().todos,
            vec![TodoItem { text: "Ship it".into(), status: "completed".into() }],
            "text is preserved from the create, status taken from the update",
        );
    }

    /// The mirror: a text-only edit must not reset a status back to pending.
    #[test]
    fn a_text_only_update_preserves_the_existing_status() {
        let mut st = BlockState::default();
        feed(&mut st, &[task_call("c1", "TaskCreate", "k1", "Old wording", "in_progress")]);
        feed(
            &mut st,
            &[format!(
                r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"u1","name":"TaskUpdate","input":{{"task_id":"k1","title":"New wording"}}}}]}}}}"#
            )],
        );
        assert_eq!(
            st.progress().todos,
            vec![TodoItem { text: "New wording".into(), status: "in_progress".into() }],
        );
    }

    /// An update for a task we never saw created (seeded mid-stream) has no
    /// text to display, so it must not invent a blank row.
    #[test]
    fn a_status_only_update_for_an_unknown_task_creates_no_row() {
        let mut st = BlockState::default();
        feed(
            &mut st,
            &[format!(
                r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"u1","name":"TaskUpdate","input":{{"task_id":"ghost","status":"completed"}}}}]}}}}"#
            )],
        );
        assert!(st.progress().todos.is_empty());
    }

    /// codex P2: Gemini-shaped transcripts put the frame at the TOP LEVEL with
    /// `tool_name`/`tool_id`/`parameters`. Before this, every non-Claude agent
    /// showed no in-flight tool and no todos at all.
    #[test]
    fn parses_provider_native_top_level_tool_frames() {
        let mut st = BlockState::default();
        feed(
            &mut st,
            &[r#"{"type":"tool_use","tool_name":"Bash","tool_id":"g1","parameters":{}}"#.to_string()],
        );
        assert_eq!(st.progress().current_tool.as_deref(), Some("Bash"));

        feed(
            &mut st,
            &[r#"{"type":"tool_result","tool_id":"g1","status":"success"}"#.to_string()],
        );
        assert_eq!(st.progress().current_tool, None, "a tool_id result clears its call");
    }

    #[test]
    fn parses_a_todo_checklist_from_a_top_level_frame() {
        let mut st = BlockState::default();
        feed(
            &mut st,
            &[r#"{"type":"tool_use","tool_name":"TodoWrite","tool_id":"g2","parameters":{"todos":[{"content":"From Gemini","status":"pending"}]}}"#.to_string()],
        );
        assert_eq!(
            st.progress().todos,
            vec![TodoItem { text: "From Gemini".into(), status: "pending".into() }],
        );
    }

    /// An ordinary transcript line must not be mistaken for a bare tool frame.
    #[test]
    fn a_plain_line_is_not_treated_as_a_tool_frame() {
        let mut st = BlockState::default();
        feed(&mut st, &[r#"{"type":"system","subtype":"init","name":"TodoWrite"}"#.to_string()]);
        assert!(st.progress().todos.is_empty());
        assert_eq!(st.progress().current_tool, None);
    }

    /// reagent P2: with no id, text is the only identity available, and keying
    /// on it alone silently merged two genuinely distinct in-flight tasks that
    /// happened to share a title. One invisibly missing row is a worse failure
    /// than two visible duplicates, so an idless CREATE always appends.
    #[test]
    fn two_idless_creates_with_the_same_title_stay_two_rows() {
        let idless_create = |id: &str, title: &str| {
            format!(
                r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"{id}","name":"TaskCreate","input":{{"title":"{title}","status":"pending"}}}}]}}}}"#
            )
        };
        let p = run(&[idless_create("c1", "Retry the flaky step"), idless_create("c2", "Retry the flaky step")]);
        assert_eq!(p.todos.len(), 2, "two distinct tasks must not collapse into one row");
        assert!(p.todos.iter().all(|t| t.text == "Retry the flaky step"));
    }

    /// …but an idless UPDATE still has to find its row, since matching by text
    /// is the only correlation available and appending would be a duplicate.
    #[test]
    fn an_idless_update_still_matches_its_row_by_text() {
        let mut st = BlockState::default();
        feed(&mut st, &[task_call("c1", "TaskCreate", "k1", "Ship it", "pending")]);
        feed(
            &mut st,
            &[format!(
                r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"u1","name":"TaskUpdate","input":{{"title":"Ship it","status":"completed"}}}}]}}}}"#
            )],
        );
        assert_eq!(
            st.progress().todos,
            vec![TodoItem { text: "Ship it".into(), status: "completed".into() }],
        );
    }

    /// An id, when present, is always the key — two same-titled tasks with
    /// distinct ids stay distinct, and an update by id finds its own row.
    #[test]
    fn ids_take_precedence_over_text_for_identity() {
        let p = run(&[
            task_call("c1", "TaskCreate", "k1", "Same title", "pending"),
            task_call("c2", "TaskCreate", "k2", "Same title", "pending"),
            task_call("u1", "TaskUpdate", "k2", "Same title", "completed"),
        ]);
        assert_eq!(
            p.todos,
            vec![
                TodoItem { text: "Same title".into(), status: "pending".into() },
                TodoItem { text: "Same title".into(), status: "completed".into() },
            ],
        );
    }

    #[test]
    fn create_style_tool_names_are_recognized() {
        assert!(is_create_call("TaskCreate"));
        assert!(is_create_call("TodoAdd"));
        assert!(!is_create_call("TaskUpdate"));
        assert!(!is_create_call("TodoWrite"), "a whole-list write never reaches the single-item path");
    }

    #[test]
    fn empty_progress_is_recognized_so_idle_agents_publish_nothing() {
        assert!(AgentProgress::default().is_empty());
        let p = run(&[tool_use("a", "Read")]);
        assert!(!p.is_empty(), "an in-flight tool is worth publishing on its own");
    }

    #[test]
    fn a_pr_create_is_recognized_in_any_shell_command() {
        assert!(is_pr_create("gh-agent pr create --title x --body y"));
        assert!(is_pr_create("cd /w/agentmux && gh pr create --fill"));
        assert!(is_pr_create("/usr/local/bin/gh-agent pr create -t x"));
        assert!(is_pr_create("C:\\npm\\gh-agent.cmd pr create -t x"));
        assert!(!is_pr_create("gh-agent pr view 12"));
        assert!(!is_pr_create("echo gh pr list; grep create"));
        assert!(!is_pr_create("git commit -m 'gh-agent pr: create'"));
        assert!(is_shell_tool("Bash") && is_shell_tool("mcp__agentmux__Shell") && !is_shell_tool("Edit"));
        assert!(is_pr_create("pwsh.exe -NoProfile -Command 'gh-agent pr create -t x'"));
    }

    /// gh's inherited flags can sit between the words, and the PR's head and
    /// base can be named.
    #[test]
    fn a_pr_create_with_flags_and_its_head_and_base() {
        assert!(is_pr_create("gh --repo owner/repo pr create -t x"));
        assert!(is_pr_create("gh pr -R owner/repo create -t x"));
        assert!(!is_pr_create("gh --repo pr view create"), "the value of --repo is not a word");
        assert!(!is_pr_create("gh pr view 1 && echo create"));
        let pr = |c: &str| pr_create(c).unwrap();
        assert_eq!(pr("gh-agent pr create -H agenty/x -B release"), NewPrCommand {
            head: Some("agenty/x".into()),
            base: Some("release".into())
        });
        assert_eq!(pr("gh pr create --head=me:feature --title t").head.as_deref(), Some("me:feature"));
        assert_eq!(pr("gh pr create -t x && git push --base y"), NewPrCommand::default(), "flags of the next command are not its");
    }

    /// Codex's shell commands are `item.started` events, as its 0.154.0
    /// fixture records them (frontend/test/fixtures/providers/codex/0.154.0/command.jsonl).
    #[test]
    fn a_codex_pr_create_is_handed_to_the_overlap_note() {
        let started = r#"{"type":"item.started","item":{"id":"item_0","type":"command_execution","command":"pwsh.exe -NoProfile -Command 'gh-agent pr create -t x'","aggregated_output":"","exit_code":null,"status":"in_progress"}}"#;
        let completed = started.replace("item.started", "item.completed");
        let mut st = BlockState::default();
        apply_lines(&mut st, &[started, completed.as_str()], Some(1_000));
        assert_eq!(st.new_prs.len(), 1, "the start counts, its completion does not");
    }

    #[test]
    fn only_a_live_pr_create_is_handed_to_the_overlap_note() {
        let line = |cmd: &str| format!(
            r#"{{"type":"assistant","message":{{"content":[{{"type":"tool_use","id":"p1","name":"Bash","input":{{"command":"{cmd}"}}}}]}}}}"#
        );
        let mut backlog = BlockState::default();
        apply_lines(&mut backlog, &[line("gh-agent pr create -t x").as_str()], None);
        assert!(backlog.new_prs.is_empty(), "backlog is history");
        let mut live = BlockState::default();
        apply_lines(&mut live, &[line("gh-agent pr create -t x").as_str(), line("ls").as_str()], Some(1_000));
        assert_eq!(live.new_prs, vec![NewPrCommand::default()]);
        assert_eq!(live.open.len(), 2, "both calls still count as in flight");
    }
}