hatchery_node/

session_tail.rs

//! Tail summary of a session's provider transcript, for a human or operator who cannot reach
//! the machine (a session went silent: what did it do last, where did it stop?).
//!
//! The session is found in the resume journal (`session-resume.json`) by its provider session
//! id (or its instance); the transcript is the provider's own file: claude
//! `~/.claude/projects/*/<id>.jsonl`, codex `~/.codex/sessions/**/rollout-*-<id>.jsonl`
//! (located and read through `codex-session-restore`). Other providers answer `unsupported`.
//!
//! The summary holds only recent prompts, tool calls with their status, errors, the MCP tools
//! the session had and where it stopped. Never tool output, env, memory or notes: text inside
//! `<system-reminder>` is dropped, a tool call naming a memory or notes path keeps only its
//! name, secrets are scrubbed and host paths shortened. Access is decided by core (owner of the
//! node or an `operator` grant on it); the node answers only core's own link.

use crate::pty_resume::Entry;
use serde::Serialize;
use serde_json::{json, Value};
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::sync::OnceLock;

/// Bytes read from the end (and the start, for the tools list) of a transcript.
const TAIL_BYTES: u64 = 2 * 1024 * 1024;
const HEAD_BYTES: u64 = 256 * 1024;
const PROMPTS: usize = 5;
const TOOL_CALLS: usize = 20;
const ERRORS: usize = 10;
const MCP_TOOLS: usize = 100;
const PROMPT_CHARS: usize = 500;
const DETAIL_CHARS: usize = 200;
const ERROR_CHARS: usize = 300;
/// A pending tool call this long without a result reads as an approval wait; past
/// [`HUNG_MS`] without any transcript write the session reads as hung.
const APPROVAL_MS: u64 = 30_000;
const HUNG_MS: u64 = 10 * 60_000;

static STATE_DIR: OnceLock<PathBuf> = OnceLock::new();

/// The node's state directory (where the resume journal lives); set once at start.
pub fn set_state_dir(dir: &Path) {
    let _ = STATE_DIR.set(dir.to_path_buf());
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Stopped {
    AwaitingApproval,
    Crashed,
    Hung,
    Finished,
    Running,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct Prompt {
    pub text: String,
    pub at: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct ToolCall {
    pub name: String,
    pub detail: String,
    /// `ok` | `error` | `pending`
    pub status: &'static str,
    pub at: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct ErrorLine {
    pub text: String,
    pub at: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct Summary {
    pub provider: String,
    pub provider_session_id: String,
    pub instance_id: u64,
    pub workspace_id: String,
    pub live: bool,
    pub stopped: Stopped,
    /// Why `stopped` reads so (read off the transcript, not the process).
    pub stopped_reason: String,
    /// Last transcript write, unix ms.
    pub last_write_ms: u64,
    pub prompts: Vec<Prompt>,
    pub tool_calls: Vec<ToolCall>,
    pub errors: Vec<ErrorLine>,
    pub mcp_tools: Vec<String>,
    /// Only the transcript's tail was read.
    pub truncated: bool,
}

/// Why no summary: `(status, message)` for core's reply.
#[derive(Debug, PartialEq)]
pub struct Refusal(pub u16, pub &'static str);

/// What the transcript said last, oldest to newest folded into one state.
#[derive(Clone, Copy, Debug, PartialEq)]
enum Last {
    Nothing,
    Prompt,
    Tool,
    Answer,
    TurnEnd,
    Aborted,
    Error,
    Approval,
}

#[derive(Default)]
struct Scan {
    prompts: Vec<Prompt>,
    tools: Vec<(String, ToolCall)>,
    errors: Vec<ErrorLine>,
    mcp: Vec<String>,
    last: Option<Last>,
}

impl Scan {
    fn prompt(&mut self, text: &str, at: Option<String>, r: &Redactor) {
        let text = r.text(&strip_reminders(text), PROMPT_CHARS);
        if !text.is_empty() {
            self.prompts.push(Prompt { text, at });
            self.last = Some(Last::Prompt);
        }
    }

    fn tool(&mut self, id: &str, name: &str, detail: String, at: Option<String>, r: &Redactor) {
        let name: String = name.chars().take(80).collect();
        if is_mcp(&name) {
            self.mcp_tool(&name);
        }
        let detail = if mentions_private(&detail) { "[memory or notes path withheld]".to_owned() } else { r.text(&detail, DETAIL_CHARS) };
        self.tools.push((id.to_owned(), ToolCall { name, detail, status: "pending", at }));
        self.last = Some(Last::Tool);
    }

    fn tool_result(&mut self, id: &str, is_error: bool) {
        if let Some((_, t)) = self.tools.iter_mut().rev().find(|(tid, _)| tid == id) {
            t.status = if is_error { "error" } else { "ok" };
        }
        self.last = Some(Last::Prompt);
    }

    fn error(&mut self, text: &str, at: Option<String>, r: &Redactor) {
        self.errors.push(ErrorLine { text: r.text(text, ERROR_CHARS), at });
        self.last = Some(Last::Error);
    }

    fn mcp_tool(&mut self, name: &str) {
        if self.mcp.len() < MCP_TOOLS && !self.mcp.iter().any(|m| m == name) {
            self.mcp.push(name.to_owned());
        }
    }
}

fn is_mcp(name: &str) -> bool {
    name.starts_with("mcp__")
}

/// The summary for the journalled session with `provider_session_id` (or `instance_id`).
pub fn tail(provider_session_id: Option<&str>, instance_id: Option<u64>) -> Result<Summary, Refusal> {
    let dir = STATE_DIR.get().ok_or(Refusal(503, "the node has no state directory"))?;
    let entries = crate::pty_resume::load(&crate::pty_resume::path(dir));
    let gone = crate::session_meta::gone_ids(dir);
    let homes = provider_homes();
    tail_in(&entries, &gone, &homes, provider_session_id, instance_id, now_ms())
}

/// [`tail`] over explicit inputs (tests).
pub fn tail_in(entries: &[Entry], gone: &[u64], homes: &[PathBuf], provider_session_id: Option<&str>, instance_id: Option<u64>, now: u64) -> Result<Summary, Refusal> {
    let entry = match (provider_session_id, instance_id) {
        (Some(sid), _) => entries.iter().find(|e| !e.provider_session_id.is_empty() && e.provider_session_id == sid),
        (None, Some(i)) => entries.iter().find(|e| e.instance_id == i),
        (None, None) => return Err(Refusal(400, "provider_session_id or instance_id is required")),
    }
    .ok_or(Refusal(404, "the session is not in the node's resume journal"))?;
    let sid = entry.provider_session_id.as_str();
    if sid.is_empty() {
        return Err(Refusal(409, "the session has no provider session id yet"));
    }
    if sid.len() > 128 || !sid.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
        return Err(Refusal(400, "bad provider session id"));
    }
    let redactor = Redactor::new(&entry.cwd, homes);
    let (scan, path, truncated) = match entry.provider.as_str() {
        "claude" => {
            let path = claude_transcript(homes, sid).ok_or(Refusal(404, "no transcript for this session"))?;
            let (head, tail, truncated) = read_ends(&path).map_err(|_| Refusal(500, "the transcript is unreadable"))?;
            (scan_claude(&head, &tail, &redactor), path, truncated)
        }
        "codex" => codex(homes, sid, &redactor)?,
        _ => return Err(Refusal(501, "unsupported: no session-restore library for this provider")),
    };
    let last_write_ms = std::fs::metadata(&path).and_then(|m| m.modified()).ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()).map_or(0, |d| d.as_millis() as u64);
    let live = !gone.contains(&entry.instance_id);
    let (stopped, stopped_reason) = stopped(&scan, live, now.saturating_sub(last_write_ms));
    let keep_last = |n: usize, len: usize| len.saturating_sub(n);
    let mut prompts = scan.prompts;
    prompts.drain(..keep_last(PROMPTS, prompts.len()));
    let mut tools: Vec<ToolCall> = scan.tools.into_iter().map(|(_, t)| t).collect();
    tools.drain(..keep_last(TOOL_CALLS, tools.len()));
    let mut errors = scan.errors;
    errors.drain(..keep_last(ERRORS, errors.len()));
    Ok(Summary {
        provider: entry.provider.clone(),
        provider_session_id: sid.to_owned(),
        instance_id: entry.instance_id,
        workspace_id: entry.workspace_id.clone(),
        live,
        stopped,
        stopped_reason,
        last_write_ms,
        prompts,
        tool_calls: tools,
        errors,
        mcp_tools: scan.mcp,
        truncated,
    })
}

fn stopped(scan: &Scan, live: bool, idle_ms: u64) -> (Stopped, String) {
    let pending = scan.tools.iter().rev().find(|(_, t)| t.status == "pending").map(|(_, t)| t.name.clone());
    let mins = idle_ms / 60_000;
    match scan.last.unwrap_or(Last::Nothing) {
        Last::Error => (Stopped::Crashed, "the last record is an error".into()),
        Last::TurnEnd | Last::Answer if !live => (Stopped::Finished, "the turn ended and the session is gone".into()),
        _ if !live => (Stopped::Crashed, "the session is gone in the middle of a turn".into()),
        Last::TurnEnd => (Stopped::Finished, "the last turn ended".into()),
        Last::Aborted => (Stopped::Finished, "the last turn was interrupted".into()),
        Last::Approval => (Stopped::AwaitingApproval, "the provider asked for an approval".into()),
        Last::Answer if idle_ms >= HUNG_MS => (Stopped::Finished, format!("no write for {mins}m after an answer")),
        Last::Nothing => (Stopped::Finished, "the transcript has no turn yet".into()),
        Last::Tool if idle_ms >= HUNG_MS => (Stopped::Hung, format!("tool call {} has no result for {mins}m", pending.unwrap_or_default())),
        Last::Tool if idle_ms >= APPROVAL_MS => (Stopped::AwaitingApproval, format!("tool call {} waits for a result", pending.unwrap_or_default())),
        _ if idle_ms >= HUNG_MS => (Stopped::Hung, format!("a turn is open and nothing was written for {mins}m")),
        _ => (Stopped::Running, "a turn is in progress".into()),
    }
}

fn scan_claude(head: &str, tail: &str, r: &Redactor) -> Scan {
    let mut scan = Scan::default();
    // The tools list is announced near the start; the tail holds the recent work.
    for line in head.lines() {
        if let Ok(v) = serde_json::from_str::<Value>(line) {
            claude_tools_listed(&mut scan, &v);
        }
    }
    let mut lines = tail.lines();
    if !tail.is_empty() && tail.len() as u64 >= TAIL_BYTES {
        lines.next(); // cut mid-record
    }
    for line in lines {
        let Ok(v) = serde_json::from_str::<Value>(line) else { continue };
        let at = v["timestamp"].as_str().map(str::to_owned);
        claude_tools_listed(&mut scan, &v);
        match v["type"].as_str() {
            Some("user") if v["isMeta"] != true => match &v["message"]["content"] {
                Value::String(s) => scan.prompt(s, at, r),
                Value::Array(blocks) => {
                    for b in blocks {
                        match b["type"].as_str() {
                            Some("text") => scan.prompt(b["text"].as_str().unwrap_or_default(), at.clone(), r),
                            Some("tool_result") => scan.tool_result(b["tool_use_id"].as_str().unwrap_or_default(), b["is_error"] == true),
                            _ => {}
                        }
                    }
                }
                _ => {}
            },
            Some("assistant") => {
                if v["isApiErrorMessage"] == true || v["error"].is_string() {
                    let text = v["message"]["content"][0]["text"].as_str().or(v["error"].as_str()).unwrap_or("provider error");
                    scan.error(text, at, r);
                    continue;
                }
                for b in v["message"]["content"].as_array().into_iter().flatten() {
                    match b["type"].as_str() {
                        Some("tool_use") => scan.tool(b["id"].as_str().unwrap_or_default(), b["name"].as_str().unwrap_or("?"), tool_detail(&b["input"]), at.clone(), r),
                        Some("text") => scan.last = Some(Last::Answer),
                        _ => {}
                    }
                }
                if v["message"]["stop_reason"] == "end_turn" {
                    scan.last = Some(Last::TurnEnd);
                }
            }
            Some("system") => match v["subtype"].as_str() {
                Some("turn_duration") => scan.last = Some(Last::TurnEnd),
                _ if v["level"] == "error" || v["subtype"] == "api_error" => {
                    let text = v["error"]["message"].as_str().or(v["content"].as_str()).unwrap_or("system error");
                    scan.error(text, at, r);
                }
                _ => {}
            },
            _ => {}
        }
    }
    scan
}

fn claude_tools_listed(scan: &mut Scan, v: &Value) {
    if v["type"] == "attachment" && v["attachment"]["type"] == "deferred_tools_delta" {
        for n in v["attachment"]["addedNames"].as_array().into_iter().flatten().filter_map(Value::as_str).filter(|n| is_mcp(n)) {
            scan.mcp_tool(n);
        }
    }
}

/// The key argument of a tool call: its command, path, pattern or URL.
fn tool_detail(input: &Value) -> String {
    let input = match input {
        Value::String(s) => serde_json::from_str(s).unwrap_or(Value::String(s.clone())),
        v => v.clone(),
    };
    if let Some(s) = input.as_str() {
        return s.to_owned();
    }
    for k in ["command", "cmd", "file_path", "path", "pattern", "url", "query", "description"] {
        match &input[k] {
            Value::String(s) => return s.clone(),
            Value::Array(a) => return a.iter().filter_map(Value::as_str).collect::<Vec<_>>().join(" "),
            _ => {}
        }
    }
    String::new()
}

#[cfg(feature = "session-restore")]
fn codex(homes: &[PathBuf], sid: &str, r: &Redactor) -> Result<(Scan, PathBuf, bool), Refusal> {
    use codex_session_restore as csr;
    let source = codex_homes(homes)
        .iter()
        .find_map(|h| csr::resolve_target(h, std::ffi::OsStr::new(sid)).ok())
        .ok_or(Refusal(404, "no transcript for this session"))?;
    let limits = csr::RestoreLimits { max_tail_bytes: TAIL_BYTES as usize, max_lines: 20_000, max_messages: 64 };
    let report = csr::load_session(&source, limits).map_err(|_| Refusal(500, "the transcript is unreadable"))?;
    let (_, tail, truncated) = read_ends(&source.path).map_err(|_| Refusal(500, "the transcript is unreadable"))?;
    let mut scan = scan_codex(&tail, r);
    // Prompts and errors as session-restore reads them (bounded, redacted), not our scan.
    scan.prompts.clear();
    scan.errors.clear();
    for m in &report.messages {
        match m.role {
            csr::MessageRole::User => scan.prompts.push(Prompt { text: r.text(&strip_reminders(&m.text), PROMPT_CHARS), at: m.timestamp.clone() }),
            csr::MessageRole::Error => scan.errors.push(ErrorLine { text: r.text(&m.text, ERROR_CHARS), at: m.timestamp.clone() }),
            _ => {}
        }
    }
    scan.prompts.retain(|p| !p.text.is_empty());
    Ok((scan, source.path, truncated || report.truncated))
}

#[cfg(not(feature = "session-restore"))]
fn codex(_: &[PathBuf], _: &str, _: &Redactor) -> Result<(Scan, PathBuf, bool), Refusal> {
    Err(Refusal(501, "unsupported: the node is built without session-restore"))
}

fn scan_codex(tail: &str, r: &Redactor) -> Scan {
    let mut scan = Scan::default();
    for line in tail.lines() {
        let Ok(v) = serde_json::from_str::<Value>(line) else { continue };
        let at = v["timestamp"].as_str().map(str::to_owned);
        let p = &v["payload"];
        match (v["type"].as_str(), p["type"].as_str()) {
            (Some("response_item"), Some("function_call" | "custom_tool_call" | "local_shell_call")) => {
                let input = p.get("arguments").or(p.get("input")).or(p.get("action")).cloned().unwrap_or(Value::Null);
                scan.tool(p["call_id"].as_str().unwrap_or_default(), p["name"].as_str().unwrap_or("shell"), tool_detail(&input), at, r);
            }
            (Some("response_item"), Some("function_call_output" | "custom_tool_call_output")) => {
                scan.tool_result(p["call_id"].as_str().unwrap_or_default(), codex_failed(&p["output"]));
            }
            (Some("response_item"), Some("message")) if p["role"] == "assistant" => scan.last = Some(Last::Answer),
            (Some("event_msg"), Some("user_message" | "task_started")) => scan.last = Some(Last::Prompt),
            (Some("event_msg"), Some("agent_message")) => scan.last = Some(Last::Answer),
            (Some("event_msg"), Some("task_complete")) => scan.last = Some(Last::TurnEnd),
            (Some("event_msg"), Some("turn_aborted")) => scan.last = Some(Last::Aborted),
            (Some("event_msg"), Some("error" | "stream_error")) => scan.error(p["message"].as_str().unwrap_or("provider error"), at, r),
            (Some("event_msg"), Some(k)) if k.ends_with("approval_request") => scan.last = Some(Last::Approval),
            (Some("event_msg"), Some("mcp_tool_call_begin" | "mcp_tool_call_end")) => {
                let inv = &p["invocation"];
                if let (Some(s), Some(t)) = (inv["server"].as_str(), inv["tool"].as_str()) {
                    scan.mcp_tool(&format!("mcp__{s}__{t}"));
                }
            }
            _ => {}
        }
    }
    scan
}

/// A codex tool output failed: a non-zero exit code in its metadata, or an error flag.
fn codex_failed(output: &Value) -> bool {
    let o = match output {
        Value::String(s) => serde_json::from_str(s).unwrap_or(Value::Null),
        v => v.clone(),
    };
    o["metadata"]["exit_code"].as_i64().is_some_and(|c| c != 0) || o["success"] == false || o["is_error"] == true
}

fn claude_transcript(homes: &[PathBuf], sid: &str) -> Option<PathBuf> {
    let name = format!("{sid}.jsonl");
    homes.iter().find_map(|h| std::fs::read_dir(h.join(".claude/projects")).ok()?.flatten().map(|d| d.path().join(&name)).find(|p| p.is_file()))
}

#[cfg(feature = "session-restore")]
fn codex_homes(homes: &[PathBuf]) -> Vec<PathBuf> {
    let mut out: Vec<PathBuf> = std::env::var_os("CODEX_HOME").filter(|v| !v.is_empty()).map(PathBuf::from).into_iter().collect();
    out.extend(homes.iter().map(|h| h.join(".codex")));
    out
}

fn provider_homes() -> Vec<PathBuf> {
    let mut homes: Vec<PathBuf> = std::env::var_os("HOME").map(PathBuf::from).into_iter().collect();
    if let Some(extra) = std::env::var_os("HATCHERY_PROVIDER_HOMES") {
        homes.extend(std::env::split_paths(&extra));
    }
    homes
}

/// The file's first [`HEAD_BYTES`] and last [`TAIL_BYTES`]; `true` when the tail is not all.
fn read_ends(path: &Path) -> std::io::Result<(String, String, bool)> {
    let mut f = std::fs::File::open(path)?;
    let len = f.metadata()?.len();
    let mut head = Vec::new();
    (&mut f).take(HEAD_BYTES.min(len)).read_to_end(&mut head)?;
    let start = len.saturating_sub(TAIL_BYTES);
    f.seek(SeekFrom::Start(start))?;
    let mut tail = Vec::new();
    f.take(TAIL_BYTES).read_to_end(&mut tail)?;
    Ok((String::from_utf8_lossy(&head).into_owned(), String::from_utf8_lossy(&tail).into_owned(), start > 0))
}

fn now_ms() -> u64 {
    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64)
}

/// Text without `<system-reminder>` blocks (memory, notes and context the harness injects).
fn strip_reminders(s: &str) -> String {
    let mut out = String::new();
    let mut rest = s;
    while let Some(i) = rest.find("<system-reminder>") {
        out.push_str(&rest[..i]);
        rest = rest[i..].find("</system-reminder>").map_or("", |j| &rest[i + j + "</system-reminder>".len()..]);
    }
    out.push_str(rest);
    out.trim().to_owned()
}

/// A path that belongs to an agent's memory or notes.
fn mentions_private(s: &str) -> bool {
    let l = s.to_ascii_lowercase();
    ["/memory/", "\\memory\\", "memory.md", "/notes/", "\\notes\\", "notes.md", "claude.md", "agents.md", "/.claude/projects/"].iter().any(|p| l.contains(p))
}

/// Scrubs secrets and host paths out of text shown off the machine.
pub struct Redactor {
    prefixes: Vec<(String, &'static str)>,
}

const SECRET_PREFIXES: &[&str] = &["sk-", "ghp_", "gho_", "ghs_", "github_pat_", "glpat-", "xoxb-", "xoxp-", "akia", "g4aho_", "eyj"];
const SECRET_KEYS: &[&str] = &["token", "secret", "password", "passwd", "api_key", "apikey", "auth", "credential", "seed", "private"];

impl Redactor {
    pub fn new(cwd: &str, homes: &[PathBuf]) -> Redactor {
        let mut prefixes: Vec<(String, &'static str)> = Vec::new();
        if cwd.len() > 1 {
            prefixes.push((cwd.trim_end_matches(['/', '\\']).to_owned(), "<workspace>"));
        }
        for h in homes {
            let h = h.to_string_lossy().trim_end_matches(['/', '\\']).to_owned();
            if h.len() > 1 {
                prefixes.push((h, "~"));
            }
        }
        prefixes.sort_by_key(|(p, _)| std::cmp::Reverse(p.len()));
        Redactor { prefixes }
    }

    /// One line of at most `max` chars, secrets and host paths replaced.
    pub fn text(&self, s: &str, max: usize) -> String {
        let mut s: String = s.split_whitespace().collect::<Vec<_>>().join(" ");
        for (p, with) in &self.prefixes {
            s = s.replace(p.as_str(), with);
        }
        let words: Vec<String> = s.split(' ').map(|w| self.word(w)).collect();
        let mut out = words.join(" ");
        if out.chars().count() > max {
            out = out.chars().take(max).collect::<String>() + "…";
        }
        out
    }

    fn word(&self, w: &str) -> String {
        // KEY=VALUE / KEY: VALUE with a secret-looking key.
        if let Some((k, v)) = w.split_once(['=', ':']).filter(|(_, v)| !v.is_empty() && !v.starts_with("//")) {
            let kl = k.to_ascii_lowercase();
            if SECRET_KEYS.iter().any(|s| kl.contains(s)) {
                return format!("{k}=[redacted]");
            }
            if looks_secret(v.trim_matches(['"', '\''])) {
                return format!("{k}=[redacted]");
            }
        }
        let bare = w.trim_matches(|c: char| "\"'`(),;[]{}".contains(c));
        if looks_secret(bare) {
            return w.replace(bare, "[redacted]");
        }
        if let Some(short) = host_path(bare) {
            return w.replace(bare, &short);
        }
        w.to_owned()
    }
}

fn looks_secret(v: &str) -> bool {
    let l = v.to_ascii_lowercase();
    if v.len() >= 12 && SECRET_PREFIXES.iter().any(|p| l.starts_with(p)) {
        return true;
    }
    // Long mixed-case base64/token-like strings (lowercase hex such as commit ids stays).
    v.len() >= 32
        && v.chars().all(|c| c.is_ascii_alphanumeric() || "+/_=-".contains(c))
        && v.chars().any(|c| c.is_ascii_uppercase())
        && v.chars().any(|c| c.is_ascii_lowercase())
        && v.chars().any(|c| c.is_ascii_digit())
}

/// An absolute host path left after the workspace/home prefixes: only its last name stays.
fn host_path(w: &str) -> Option<String> {
    let windows = w.len() > 3 && w.as_bytes()[1] == b':' && (w.as_bytes()[2] == b'\\' || w.as_bytes()[2] == b'/');
    if !(w.starts_with('/') && w[1..].contains('/')) && !windows {
        return None;
    }
    let last = w.trim_end_matches(['/', '\\']).rsplit(['/', '\\']).next().unwrap_or_default();
    Some(format!("<host>/{last}"))
}

/// The `{"ok", "body"|"status","message"}` reply for core.
pub fn reply(req: &Value) -> Value {
    let sid = req["provider_session_id"].as_str();
    let inst = req["instance_id"].as_u64();
    match tail(sid, inst) {
        Ok(s) => json!({ "ok": true, "body": s }),
        Err(Refusal(status, message)) => json!({ "ok": false, "status": status, "message": message }),
    }
}

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

    const SID: &str = "0b9c3a52-1111-4222-8333-944455556666";

    fn entry(provider: &str, inst: u64, sid: &str, cwd: &str) -> Entry {
        Entry { instance_id: inst, provider: provider.into(), workspace_id: "task-7".into(), provider_session_id: sid.into(), cwd: cwd.into(), acp: true, ..Default::default() }
    }

    fn write(path: &Path, lines: &[Value]) {
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let mut f = std::fs::File::create(path).unwrap();
        for l in lines {
            writeln!(f, "{l}").unwrap();
        }
    }

    fn claude_home(lines: &[Value]) -> tempfile::TempDir {
        let home = tempfile::tempdir().unwrap();
        write(&home.path().join(".claude/projects/-w/").join(format!("{SID}.jsonl")), lines);
        home
    }

    fn tool_use(id: &str, name: &str, input: Value) -> Value {
        json!({"type": "assistant", "timestamp": "t2", "message": {"content": [{"type": "tool_use", "id": id, "name": name, "input": input}]}})
    }

    fn tool_result(id: &str, is_error: bool) -> Value {
        json!({"type": "user", "message": {"content": [{"type": "tool_result", "tool_use_id": id, "is_error": is_error, "content": "SECRET OUTPUT"}]}})
    }

    #[test]
    fn claude_tail_reads_prompts_tools_errors_and_mcp_without_private_text() {
        let home = claude_home(&[
            json!({"type": "attachment", "attachment": {"type": "deferred_tools_delta", "addedNames": ["Bash", "mcp__m4a__send", "mcp__m4a__inbox"]}}),
            json!({"type": "user", "timestamp": "t1", "message": {"content": "<system-reminder>memory: the owner's notes</system-reminder>fix the build in /w/task-7/src"}}),
            json!({"type": "user", "isMeta": true, "message": {"content": "meta"}}),
            tool_use("a", "Bash", json!({"command": "cargo test GITHUB_TOKEN=ghp_abcdefghijklmnop1234"})),
            tool_result("a", true),
            tool_use("b", "Read", json!({"file_path": "/home/u/.claude/projects/x/memory/MEMORY.md"})),
            tool_result("b", false),
            tool_use("c", "mcp__m4a__send", json!({"to": "hatcheryweb"})),
            json!({"type": "system", "level": "error", "subtype": "api_error", "error": {"message": "overloaded at /opt/secret/dir/x.log"}}),
        ]);
        let entries = [entry("claude", 3, SID, "/w/task-7")];
        let s = tail_in(&entries, &[], &[home.path().to_path_buf()], Some(SID), None, now_ms()).unwrap();
        assert_eq!(s.prompts, vec![Prompt { text: "fix the build in <workspace>/src".into(), at: Some("t1".into()) }]);
        let tools: Vec<_> = s.tool_calls.iter().map(|t| (t.name.as_str(), t.status, t.detail.as_str())).collect();
        assert_eq!(tools[0], ("Bash", "error", "cargo test GITHUB_TOKEN=[redacted]"));
        assert_eq!(tools[1], ("Read", "ok", "[memory or notes path withheld]"));
        assert_eq!((tools[2].0, tools[2].1), ("mcp__m4a__send", "pending"));
        assert_eq!(s.errors[0].text, "overloaded at <host>/x.log");
        assert_eq!(s.mcp_tools, ["mcp__m4a__send", "mcp__m4a__inbox"]);
        assert_eq!((s.stopped, s.live, s.instance_id), (Stopped::Crashed, true, 3));
        let all = serde_json::to_string(&s).unwrap();
        for leak in ["SECRET OUTPUT", "owner's notes", "ghp_", "/opt/secret", "MEMORY.md"] {
            assert!(!all.contains(leak), "{leak} leaked: {all}");
        }
    }

    #[test]
    fn where_it_stopped() {
        let sc = |last: Last, pending: bool| {
            let mut s = Scan { last: Some(last), ..Default::default() };
            if pending {
                s.tools.push(("x".into(), ToolCall { name: "Bash".into(), detail: String::new(), status: "pending", at: None }));
            }
            s
        };
        let st = |s: &Scan, live: bool, idle: u64| stopped(s, live, idle).0;
        assert_eq!(st(&sc(Last::TurnEnd, false), true, 5), Stopped::Finished);
        assert_eq!(st(&sc(Last::TurnEnd, false), false, 5), Stopped::Finished);
        assert_eq!(st(&sc(Last::Tool, true), false, 5), Stopped::Crashed, "gone mid-turn");
        assert_eq!(st(&sc(Last::Error, false), true, 5), Stopped::Crashed);
        assert_eq!(st(&sc(Last::Tool, true), true, 5_000), Stopped::Running);
        assert_eq!(st(&sc(Last::Tool, true), true, 60_000), Stopped::AwaitingApproval);
        assert_eq!(st(&sc(Last::Approval, false), true, 1), Stopped::AwaitingApproval);
        assert_eq!(st(&sc(Last::Tool, true), true, HUNG_MS), Stopped::Hung);
        assert_eq!(st(&sc(Last::Prompt, false), true, HUNG_MS + 1), Stopped::Hung);
        assert_eq!(st(&sc(Last::Prompt, false), true, 10), Stopped::Running);
        assert!(stopped(&sc(Last::Tool, true), true, HUNG_MS).1.contains("Bash"));
    }

    #[test]
    fn found_only_through_the_resume_journal() {
        let home = claude_home(&[json!({"type": "user", "message": {"content": "hi"}})]);
        let homes = [home.path().to_path_buf()];
        let entries = [entry("claude", 3, SID, "/w"), entry("grok", 4, "g-1", "/w"), entry("claude", 5, "", "/w"), entry("claude", 6, "../../etc/passwd", "/w")];
        let t = |sid: Option<&str>, inst: Option<u64>| tail_in(&entries, &[3], &homes, sid, inst, 0);
        let by_instance = t(None, Some(3)).unwrap();
        assert_eq!((by_instance.provider_session_id.as_str(), by_instance.live), (SID, false));
        assert_eq!(t(Some("not-journalled"), None).unwrap_err(), Refusal(404, "the session is not in the node's resume journal"));
        assert_eq!(t(None, None).unwrap_err().0, 400);
        assert_eq!(t(Some("g-1"), None).unwrap_err().0, 501, "grok: unsupported");
        assert_eq!(t(None, Some(5)).unwrap_err().0, 409);
        assert_eq!(t(None, Some(6)).unwrap_err().0, 400, "no path tricks");
    }

    #[test]
    fn redaction_keeps_commit_ids_and_drops_secrets_and_paths() {
        let r = Redactor::new("/srv/ws/task-7", &[PathBuf::from("/home/u")]);
        assert_eq!(r.text("git show 0123456789abcdef0123456789abcdef01234567", 200), "git show 0123456789abcdef0123456789abcdef01234567");
        assert_eq!(r.text("export ANTHROPIC_API_KEY=xyz", 200), "export ANTHROPIC_API_KEY=[redacted]");
        assert_eq!(r.text("Authorization: Bearer abcDEF0123456789abcDEF0123456789xyz", 200), "Authorization: Bearer [redacted]");
        assert_eq!(r.text("cat /srv/ws/task-7/a.rs ~/x /home/u/.codex/config.toml /etc/hosts", 200), "cat <workspace>/a.rs ~/x ~/.codex/config.toml <host>/hosts");
        assert_eq!(r.text("see https://example.com/a/b", 200), "see https://example.com/a/b");
        assert_eq!(r.text(&"x".repeat(50), 10), format!("{}…", "x".repeat(10)));
        assert_eq!(strip_reminders("a<system-reminder>b</system-reminder>c<system-reminder>open"), "ac");
    }

    #[cfg(feature = "session-restore")]
    #[test]
    fn codex_tail_through_session_restore() {
        let home = tempfile::tempdir().unwrap();
        let rollout = home.path().join(".codex/sessions/2026/10/19").join(format!("rollout-2026-10-19T10-00-00-{SID}.jsonl"));
        let rec = |ty: &str, payload: Value| json!({"timestamp": "2026-10-19T10:00:01Z", "type": ty, "payload": payload});
        write(&rollout, &[
            rec("session_meta", json!({"id": SID, "cwd": "/w/task-7", "model_provider": "openai"})),
            rec("event_msg", json!({"type": "user_message", "message": "why is ci red"})),
            rec("response_item", json!({"type": "message", "role": "user", "content": [{"type": "input_text", "text": "why is ci red"}]})),
            rec("response_item", json!({"type": "function_call", "name": "shell", "arguments": "{\"command\":[\"bash\",\"-lc\",\"cargo test\"]}", "call_id": "c1"})),
            rec("response_item", json!({"type": "function_call_output", "call_id": "c1", "output": "{\"output\":\"boom\",\"metadata\":{\"exit_code\":101}}"})),
            rec("event_msg", json!({"type": "mcp_tool_call_end", "invocation": {"server": "m4a", "tool": "send"}})),
            rec("response_item", json!({"type": "function_call", "name": "shell", "arguments": "{\"command\":[\"bash\",\"-lc\",\"git push\"]}", "call_id": "c2"})),
            rec("event_msg", json!({"type": "exec_approval_request", "call_id": "c2"})),
        ]);
        let entries = [entry("codex", 21, SID, "/w/task-7")];
        let s = tail_in(&entries, &[], &[home.path().to_path_buf()], Some(SID), None, now_ms()).unwrap();
        assert_eq!(s.stopped, Stopped::AwaitingApproval);
        let tools: Vec<_> = s.tool_calls.iter().map(|t| (t.detail.as_str(), t.status)).collect();
        assert_eq!(tools, [("bash -lc cargo test", "error"), ("bash -lc git push", "pending")]);
        assert_eq!(s.mcp_tools, ["mcp__m4a__send"]);
        assert!(s.prompts.iter().any(|p| p.text == "why is ci red"), "{:?}", s.prompts);
        assert!(!serde_json::to_string(&s).unwrap().contains("boom"), "no tool output");
    }
}