crates/memory/src/work.rs

//! Non-interactive runs: a prompt, or a task claimed from the ledger.
//!
//! The launcher reaches the ledger and the record the way a harness does, through the
//! memory server's tool surface, over an in-process pipe. The ledger's rules stay on the
//! server; this module knows how to turn a task into a prompt and an exit into a handoff.

use crate::hooks::{self, Event};
use crate::Session;
use dm_client::{MemoryClient, Task};
use dm_sandbox::Harness;
use dm_server::{mcp, prefetch, Memory};
use std::collections::BTreeSet;
use std::sync::Arc;
use std::time::Duration;

type Fallible<T> = Result<T, Box<dyn std::error::Error>>;

/// Exit code: nothing was claimable.
pub const NOTHING_TO_DO: i32 = 3;
/// Exit code: the harness stopped without completing, and the task went back.
pub const STOPPED: i32 = 4;

/// How `work` runs.
pub struct Options {
    /// Claim from this domain; the only granted domain when omitted.
    pub domain: Option<String>,
    /// Lease length; heartbeats come every third of it.
    pub lease_seconds: i64,
    /// Keep claiming until interrupted.
    pub keep_going: bool,
    /// How long to wait when the queue is empty, with `keep_going`.
    pub idle: Duration,
    /// Mine the transcript after each task, when set.
    pub mine: Option<MinePlan>,
    /// Put the summaries of the entries a task likely needs in its prompt; off for the
    /// no-memory comparisons.
    pub prefetch: bool,
    /// Claim this task rather than the next one, and stop after it.
    pub task: Option<String>,
}

/// How to mine after a task.
pub struct MinePlan {
    /// The model, or the destination's small default.
    pub model: Option<String>,
    /// Through a prompted run of the harness rather than a direct call.
    pub via_harness: bool,
}

/// A client joined to this process's memory server by a byte pipe: the same JSON-RPC
/// a harness speaks, with no socket between.
///
/// # Errors
///
/// When the MCP handshake fails.
pub async fn client(memory: Arc<Memory>) -> Fallible<MemoryClient> {
    let (client_io, server_io) = tokio::io::duplex(256 * 1024);
    let (sr, sw) = tokio::io::split(server_io);
    tokio::spawn(async move {
        if let Err(e) = mcp::serve_launcher_pipe(memory, sr, sw).await {
            eprintln!("amem: in-process server: {e}");
        }
    });
    let (cr, cw) = tokio::io::split(client_io);
    Ok(MemoryClient::connect((cr, cw)).await?)
}

async fn handoff_ids(client: &MemoryClient, domain: Option<&str>) -> Fallible<BTreeSet<String>> {
    Ok(client
        .handoffs(domain, 100)
        .await?
        .into_iter()
        .map(|(id, _)| id)
        .collect())
}

/// File what the harness's hooks left in the outbox. The transcript the run left is
/// kept in the record first, as a source in `domain`, the session's, so each note cites
/// the turns it was drawn from. A note from before compaction is always filed: what it
/// holds is gone from the harness's context. A note from session end is filed only when
/// the run left no handoff of its own, as the handoff of last resort; otherwise it is
/// dropped. Returns the ids filed.
///
/// # Errors
///
/// When the outbox cannot be read or the memory server refuses a write.
pub async fn file_outbox(
    session: &Session,
    client: &MemoryClient,
    wrote_handoff: bool,
    domain: Option<&str>,
) -> Fallible<Vec<String>> {
    let memory = session.memory();
    let kept = crate::transcripts::keep_newest(
        &memory,
        session.harness().binary(),
        domain,
        session.home(),
    )
    .await;
    let mut filed = Vec::new();
    for note in hooks::notes(&session.outbox())? {
        if note.event == Event::SessionEnd && wrote_handoff {
            std::fs::remove_file(&note.path).ok();
            continue;
        }
        let source = kept
            .as_ref()
            .map(|k| dm_server::transcripts::cite(&k.session, &note.turns))
            .unwrap_or_default();
        let id = client
            .write_entry(&dm_client::NewEntry {
                kind: "handoff",
                title: &note.title,
                body: &note.body,
                confidence: None,
                source,
                domain,
            })
            .await?;
        eprintln!(
            "amem: handoff {id} from the harness's hook ({})",
            note.event.suffix()
        );
        std::fs::remove_file(&note.path).ok();
        filed.push(id);
    }
    Ok(filed)
}

/// An interactive run: the harness as the person knows it, with the outbox filed when
/// it exits.
///
/// # Errors
///
/// When the harness cannot be run or the memory server cannot be reached.
pub async fn interactive(session: &Session, args: &[String]) -> Fallible<i32> {
    let client = client(session.memory()).await?;
    let before = handoff_ids(&client, None).await?;
    let (code, _) = session.run(args, false).await?;
    let after = handoff_ids(&client, None).await?;
    file_outbox(
        session,
        &client,
        after.difference(&before).next().is_some(),
        None,
    )
    .await?;
    client.close().await.ok();
    Ok(code)
}

/// Run one prompt in the harness's headless form. Unless `no_handoff`, a run that left
/// no handoff exits [`STOPPED`]: a non-interactive run whose outcome is only in its
/// transcript has left nothing for the next session.
///
/// # Errors
///
/// When the harness cannot be run or the memory server cannot be reached.
pub async fn prompted(
    session: &Session,
    prompt: &str,
    args: &[String],
    no_handoff: bool,
) -> Fallible<i32> {
    let client = client(session.memory()).await?;
    let before = handoff_ids(&client, None).await?;
    // the head of the session's tape: what a re-enactment starts the same way from
    let commit = crate::reenact::commit_of(session.workdir());
    session
        .memory()
        .tape_start(dm_server::tape::TapeStart {
            harness: session.harness().binary().to_string(),
            model: session.model().map(str::to_string),
            prompt: Some(prompt.to_string()),
            dirty: commit.as_ref().is_some_and(|(_, d)| *d),
            commit: commit.map(|(c, _)| c),
        })
        .await
        .ok();
    // the harness's own arguments follow its headless form, where every harness takes
    // a flag after the prompt
    let mut argv = session.harness().headless(prompt);
    argv.extend(args.iter().cloned());
    let (code, output) = session.run(&argv, true).await?;
    let output = output.map(|o| answer_of(session, o));
    if let Some(out) = &output {
        crate::stops::answer(out);
    }
    let mut after = handoff_ids(&client, None).await?;
    let wrote = after.difference(&before).next().is_some();
    after.extend(file_outbox(session, &client, wrote, None).await?);
    let new: Vec<&String> = after.difference(&before).collect();
    if new.is_empty() && !no_handoff {
        eprintln!(
            "amem: the run left no handoff; the next session starts from nothing (pass --no-handoff if that is intended)"
        );
        client.close().await.ok();
        return Ok(if code == 0 { STOPPED } else { code });
    }
    for id in new {
        eprintln!("amem: handoff {id}");
    }
    client.close().await.ok();
    Ok(code)
}

/// What a headless run printed, as the answer. Muse Code's `exec --json` prints its
/// record stream, one JSON object a line: that stream is kept under the private home
/// as the run's transcript, since Muse's own session store is binary, and the answer is
/// the text of its `run.terminal.completed` record. Every other harness prints the
/// answer itself.
fn answer_of(session: &Session, output: String) -> String {
    if session.harness() != Harness::Muse {
        return output;
    }
    let dir = session.home().join(".local/share/muse/exec");
    if std::fs::create_dir_all(&dir).is_ok() {
        let stamp = time::OffsetDateTime::now_utc().unix_timestamp_nanos();
        std::fs::write(dir.join(format!("{stamp}.jsonl")), &output).ok();
    }
    muse_answer(&output).unwrap_or(output)
}

/// The final text of a Muse Code record stream, when it holds one: what the run said
/// when it completed, or why it failed.
fn muse_answer(jsonl: &str) -> Option<String> {
    jsonl
        .lines()
        .filter_map(|l| serde_json::from_str::<serde_json::Value>(l).ok())
        .filter_map(|v| {
            let text = match v.get("payload_type").and_then(serde_json::Value::as_str)? {
                "run.terminal.completed" => v.pointer("/payload/text")?.as_str()?.to_string(),
                "run.terminal.failed" => {
                    let reason = v
                        .pointer("/payload/reason")
                        .and_then(serde_json::Value::as_str);
                    format!(
                        "muse: the run failed: {}",
                        reason.unwrap_or("no reason given")
                    )
                }
                _ => return None,
            };
            Some(text)
        })
        .next_back()
}

/// Claim and run tasks.
///
/// # Errors
///
/// When the harness cannot be run or the memory server cannot be reached.
pub async fn run(session: &Session, options: &Options) -> Fallible<i32> {
    let client = Arc::new(client(session.memory()).await?);
    let mut last = NOTHING_TO_DO;
    loop {
        // counted from before the claim until the task is settled: a launcher told to
        // stop waits, and the task goes back to the pool with a handoff, as when the
        // harness stops of itself. One told to stop claims nothing more.
        let Ok(in_flight) = crate::stops::InFlight::enter() else {
            return Ok(last);
        };
        let claimed = match &options.task {
            Some(id) => client.claim(id, options.lease_seconds).await?,
            // a worker only claims: rank is the ledger's, moved along dependency edges
            // at admission and by triage's cycle, never by a worker on its way in
            None => {
                client
                    .claim_next(options.domain.as_deref(), options.lease_seconds)
                    .await?
            }
        };
        let Some(task) = claimed else {
            drop(in_flight);
            if !options.keep_going {
                eprintln!("amem: nothing claimable");
                return Ok(if last == NOTHING_TO_DO {
                    NOTHING_TO_DO
                } else {
                    last
                });
            }
            tokio::time::sleep(options.idle).await;
            continue;
        };
        if dm_sandbox::stop::stopping() {
            // claimed while the stop came: not started, so back with no handoff
            client.release(&task.id, None).await.ok();
            return Ok(last);
        }
        last = one(session, &client, &task, options).await?;
        drop(in_flight);
        if options.task.is_some() {
            return Ok(last);
        }
        if let Some(plan) = options
            .mine
            .as_ref()
            .filter(|_| !dm_sandbox::stop::stopping())
        {
            let via = if plan.via_harness {
                crate::MineVia::Harness
            } else {
                crate::MineVia::Api
            };
            if let Err(e) = session.mine(plan.model.as_deref(), via).await {
                eprintln!("amem: mining failed: {e}");
            }
        }
        if !options.keep_going {
            return Ok(last);
        }
    }
}

async fn one(
    session: &Session,
    client: &Arc<MemoryClient>,
    task: &Task,
    options: &Options,
) -> Fallible<i32> {
    let lease_seconds = options.lease_seconds;
    eprintln!("amem: claimed {}: {}", task.id, task.title);
    let holding = Holding::of(&task.id);
    let memory = session.memory();
    let platform = Platform::for_harness(&memory, session.harness()).await;
    let prompt = prompt_for(
        client,
        task,
        options.prefetch.then_some(&*memory),
        &platform,
    )
    .await?;
    // the reads above were the launcher's: neither the fan-out rule nor the pushed set
    // should carry them into the harness's session
    session.memory().harness_starts();
    let heartbeat = {
        let client = client.clone();
        let id = task.id.clone();
        tokio::spawn(async move {
            let every = Duration::from_secs((lease_seconds / 3).max(5).unsigned_abs());
            let mut tick = tokio::time::interval(every);
            tick.tick().await;
            loop {
                tick.tick().await;
                if let Err(e) = client.heartbeat(&id, lease_seconds).await {
                    eprintln!("amem: heartbeat on {id}: {e}");
                    return;
                }
            }
        })
    };
    let run = session
        .run_task(&session.harness().headless(&prompt), true, Some(&task.id))
        .await;
    heartbeat.abort();
    let (code, output) = match run {
        Ok(ran) => ran,
        Err(e) => {
            // a run refused because a stop came first leaves nothing to settle
            if dm_sandbox::stop::stopping() {
                client.release(&task.id, None).await.ok();
            }
            return Err(e);
        }
    };
    let output = output.map(|o| answer_of(session, o));
    let now = client.get_task(&task.id).await?;
    file_outbox(
        session,
        client,
        now.status == "completed",
        Some(&task.domain),
    )
    .await?;
    let output = output.unwrap_or_default();
    if !output.trim().is_empty() {
        crate::stops::answer(&output);
    }
    // settled from here: a launcher told twice to stop no longer releases it, as the
    // release or completion below may already be done, and the task another's
    drop(holding);
    settle(session.harness(), client, task, code, output).await
}

/// The task this worker holds, if any: what a launcher told twice to stop releases
/// before it exits.
static HOLDING: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);

/// The task this worker holds, while it holds one.
pub fn holding() -> Option<String> {
    HOLDING
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone()
}

/// A task held, from its claim until it is settled.
struct Holding;

impl Holding {
    fn of(id: &str) -> Holding {
        *HOLDING
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(id.to_string());
        Holding
    }
}

impl Drop for Holding {
    fn drop(&mut self) {
        *HOLDING
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
    }
}

/// After the harness exits: a completed task stays complete; a task still held goes
/// back to the pool with the harness's output filed as a handoff.
async fn settle(
    harness: Harness,
    client: &MemoryClient,
    task: &Task,
    code: i32,
    output: String,
) -> Fallible<i32> {
    let now = client.get_task(&task.id).await?;
    match now.status.as_str() {
        "completed" => {
            eprintln!(
                "amem: {} completed, handoff {}",
                task.id,
                now.handoff.as_deref().unwrap_or("-")
            );
            Ok(0)
        }
        "claimed" if delegated_report(task, code, &output) => {
            let handoff = client
                .write_handoff(
                    &format!("Report: {}", task.title),
                    &format!(
                        "# {title}\n\nThe report `{harness}` returned for task {id}, filed by the board for a run that delegated it. A delegated task's deliverable is its report, as a subagent's is to the session that spawned it, so the task is complete with this as its handoff; the continuation that waits for it reads it.\n\n## Report\n\n{tail}\n",
                        title = task.title,
                        harness = harness.binary(),
                        id = task.id,
                        tail = tail_of(&output, 6000),
                    ),
                    Some(&task.domain),
                )
                .await?;
            client.complete(&task.id, &handoff).await?;
            eprintln!(
                "amem: {} completed with its report as handoff {handoff}",
                task.id
            );
            Ok(0)
        }
        "claimed" => {
            let tail = tail_of(&output, 6000);
            let body = format!(
                "# {title}\n\n`{harness}` stopped without completing task {id} (exit {code}). \
                 The task is back in the pool; what follows is what it printed, so the next \
                 claimant can start from here.\n\n## Output\n\n{tail}\n",
                title = task.title,
                harness = harness.binary(),
                id = task.id,
            );
            let handoff = client
                .write_handoff(
                    &format!("Task {} stopped: {}", task.id, task.title),
                    &body,
                    Some(&task.domain),
                )
                .await?;
            client.release(&task.id, Some(&handoff)).await?;
            eprintln!(
                "amem: {} released; its output is filed as handoff {handoff}",
                task.id
            );
            Ok(STOPPED)
        }
        other => {
            eprintln!("amem: {} is now {other}", task.id);
            Ok(if code == 0 { STOPPED } else { code })
        }
    }
}

/// Whether a run's output is a delegated task's report: a task the board filed for a
/// run that delegated it (`origin: task:<parent>`), which ended cleanly with something
/// to report. Such a task is done when it reports, whether or not the subagent's own
/// prompt let it call `complete`; anything else that stopped goes back to the pool.
fn delegated_report(task: &Task, code: i32, output: &str) -> bool {
    code == 0
        && !output.trim().is_empty()
        && task
            .raw
            .get("origin")
            .and_then(serde_json::Value::as_str)
            .is_some_and(|o| o.starts_with("task:"))
}

fn tail_of(s: &str, chars: usize) -> String {
    let s = s.trim();
    if s.is_empty() {
        return "(nothing)".into();
    }
    let n = s.chars().count();
    if n <= chars {
        return s.to_string();
    }
    let skip = n - chars;
    format!("…\n{}", s.chars().skip(skip).collect::<String>())
}

/// The prompt: the task, what the tasks it depends on left behind, the summaries of what
/// it likely needs when `prefetch_from` gives the session's memory, T0 for its domain, and how
/// to stop.
/// What the platform's domain gives a task's prompt: verified artifacts only.
#[derive(Default)]
pub struct Platform {
    /// The opening, a `template` named `work`, in place of the launcher's own.
    template: Option<dm_server::platform::Artifact>,
    /// An `agent` definition for this harness, or the `default` one.
    agent: Option<dm_server::platform::Artifact>,
}

impl Platform {
    /// The verified artifacts for `harness`; none where the platform holds none.
    pub async fn for_harness(memory: &Memory, harness: Harness) -> Platform {
        let get = |kind: &'static str, name: String| async move {
            memory.platform_artifact(kind, &name).await.ok().flatten()
        };
        let agent = match get("agent", harness.binary().to_string()).await {
            Some(a) => Some(a),
            None => get("agent", "default".to_string()).await,
        };
        Platform {
            template: get("template", "work".to_string()).await,
            agent,
        }
    }
}

impl Platform {
    /// The prompt's opening: the platform's template or the launcher's own, then the
    /// agent definition, each said to be the platform's.
    fn opening(&self) -> String {
        let mut p = match &self.template {
            Some(t) => t.text.trim().to_string(),
            None => OPENING.to_string(),
        };
        p.push_str("\n\n");
        if let Some(a) = &self.agent {
            p.push_str(&format!(
                "# How to work here ({}, verified)\n\n{}\n\n",
                a.entry,
                a.text.trim()
            ));
        }
        p
    }
}

/// The launcher's own opening of a task's prompt, when the platform gives none.
const OPENING: &str =
    "You are working one task from a shared task ledger, reached through the `memory` MCP server. \
     The launcher holds the lease and renews it for you; do not call `heartbeat`. \
     Your transcript is kept as a source, and your provider may keep your reasoning out of it, \
     so before each decision that changes something, choosing between entries, writing, \
     completing or releasing, say in one plain sentence why.";

async fn prompt_for(
    client: &MemoryClient,
    task: &Task,
    prefetch_from: Option<&Memory>,
    platform: &Platform,
) -> Fallible<String> {
    let mut p = platform.opening();
    p.push_str(&format!(
        "# Task {}: {}\n\n{}\n",
        task.id,
        task.title,
        task.body.trim()
    ));
    if !task.requirements.is_empty() {
        p.push_str(&format!("\nRequires: {}\n", task.requirements.join(", ")));
    }
    if let Some(h) = &task.handoff {
        if let Ok(e) = client.get(h).await {
            p.push_str(&format!(
                "\n# Where the last attempt stopped ({h})\n\n{}\n",
                e.body.trim()
            ));
        }
    }
    if !task.depends_on.is_empty() {
        p.push_str("\n# What the tasks this one depends on left behind\n");
        p.push_str(
            "\nYour first act is to verify these against what this task assumes: an entry's `review_by` may have passed, a decision may have been superseded, a handoff may describe a world that has moved on. Check what matters with `get` and `search`. A mismatch is not absorbed silently: say so in your handoff and release the task, so the board can re-decompose it.\n",
        );
        for dep in &task.depends_on {
            let Ok(d) = client.get_task(dep).await else {
                continue;
            };
            match &d.handoff {
                Some(h) => match client.get(h).await {
                    Ok(e) => p.push_str(&format!(
                        "\n## {dep}: {} → handoff {h}\n\n{}\n",
                        d.title,
                        tail_of(&e.body, 4000)
                    )),
                    Err(_) => p.push_str(&format!(
                        "\n## {dep}: {} → handoff {h} (unreadable here)\n",
                        d.title
                    )),
                },
                None => p.push_str(&format!("\n## {dep}: {} ({})\n", d.title, d.status)),
            }
        }
    }
    if let Some(memory) = prefetch_from {
        p.push_str(&prefetched(memory, task).await);
    }
    let t0 = client.t0(Some(&task.domain)).await?;
    p.push_str(&format!("\n# T0 for {}\n\n{}\n", task.domain, t0.trim()));
    p.push_str(&format!(
        "\n# When you stop\n\nWrite a `handoff` entry with the `write` tool (entry_type handoff): what you did \
         and why, what is left, where to look. Then call `complete` with task id `{id}` and the handoff's id. \
         If you cannot finish, still write the handoff and stop; do not call `complete`, and the task \
         returns to the pool with your handoff attached.\n",
        id = task.id
    ));
    Ok(p)
}

/// The prefetch for a task, as its prompt section. A prefetch that fails costs the
/// task nothing but the section: the harness can still search.
async fn prefetched(memory: &Memory, task: &Task) -> String {
    let query = prefetch::task_query(&task.title, &task.body);
    match memory.prefetch(&query, &prefetch::Plan::default()).await {
        Ok(found) => {
            eprintln!(
                "amem: prefetched {} summaries ({} tokens): {}",
                found.hits.len(),
                found.tokens,
                found.ids().join(", ")
            );
            prefetch_section(&found)
        }
        Err(e) => {
            eprintln!("amem: no prefetch for {}: {e}", task.id);
            String::new()
        }
    }
}

/// The prefetched summaries as a prompt section; nothing when nothing was found.
fn prefetch_section(found: &prefetch::Prefetched) -> String {
    if found.hits.is_empty() {
        return String::new();
    }
    let mut s = String::from(
        "\n# Entries this task likely needs: summaries only\n\n\
         Chosen before you started, by a search on the task and by what earlier sessions \
         retrieved together. These are summaries, not bodies: open one with `get` when you \
         need what it says, and leave the rest.\n\n",
    );
    for h in &found.hits {
        let summary = h.summary.trim();
        s.push_str(&format!(
            "- `{}` ({}, {}): {}",
            h.id,
            h.entry_type,
            h.domain,
            h.title.trim()
        ));
        if !summary.is_empty() && summary != h.title.trim() {
            s.push_str(&format!(". {summary}"));
        }
        if let Some((anchor, n)) = found.with.get(&h.id) {
            s.push_str(&format!(" (retrieved with `{anchor}` {n} times)"));
        }
        s.push('\n');
    }
    s
}

#[cfg(test)]
mod tests {

    fn artifact(kind: &str, text: &str) -> dm_server::platform::Artifact {
        dm_server::platform::Artifact {
            kind: kind.into(),
            name: "x".into(),
            text: text.into(),
            entry: format!("mem-plat-{kind}"),
            created: String::new(),
        }
    }

    #[test]
    fn the_platform_s_template_and_agent_open_the_prompt_and_say_whose_they_are() {
        let none = Platform::default().opening();
        assert!(none.starts_with(OPENING), "{none}");
        let p = Platform {
            template: Some(artifact(
                "template",
                "Work the task; say why before each write.",
            )),
            agent: Some(artifact(
                "agent",
                "Read the decision entry before a migration.",
            )),
        }
        .opening();
        assert!(
            p.starts_with("Work the task; say why before each write."),
            "{p}"
        );
        assert!(!p.contains(OPENING));
        assert!(
            p.contains("# How to work here (mem-plat-agent, verified)"),
            "{p}"
        );
        assert!(p.contains("before a migration"));
    }

    #[test]
    fn a_delegated_tasks_clean_report_completes_it() {
        let task = |origin: Option<&str>| Task {
            id: "task-a-1".into(),
            title: "t".into(),
            body: "b".into(),
            domain: "dom-a".into(),
            status: "claimed".into(),
            depends_on: Vec::new(),
            requirements: Vec::new(),
            handoff: None,
            raw: serde_json::json!({ "origin": origin }),
        };
        assert!(delegated_report(
            &task(Some("task:task-a-0")),
            0,
            "DR-19 is proposed."
        ));
        assert!(
            !delegated_report(&task(Some("task:task-a-0")), 1, "failed"),
            "a failed run is not a report"
        );
        assert!(
            !delegated_report(&task(Some("task:task-a-0")), 0, "  "),
            "nothing to report"
        );
        assert!(
            !delegated_report(&task(Some("librarian:merge")), 0, "done"),
            "only delegated tasks"
        );
        assert!(!delegated_report(&task(None), 0, "done"));
    }

    use super::*;

    #[test]
    fn the_prefetch_section_lists_summaries_and_says_how_to_open_them() {
        assert!(prefetch_section(&prefetch::Prefetched::default()).is_empty());
        let hit = |id: &str, title: &str, summary: &str| dm_index::Hit {
            id: id.into(),
            domain: "dom-demo".into(),
            entry_type: dm_core::EntryType::Fact,
            trust_tier: dm_core::TrustTier::Raw,
            classification: dm_core::Label::Public,
            title: title.into(),
            summary: summary.into(),
            score: 0.0,
            review_due: false,
        };
        let found = prefetch::Prefetched {
            query: "retry".into(),
            hits: vec![
                hit(
                    "mem-a-1",
                    "Retry backoff",
                    "Backoff doubles, capped at 15 minutes.",
                ),
                hit("mem-a-2", "Context header", "Context header"),
            ],
            with: [("mem-a-2".to_string(), ("mem-a-1".to_string(), 4))].into(),
            tokens: 40,
        };
        let s = prefetch_section(&found);
        assert!(s.contains("summaries only") && s.contains("`get`"), "{s}");
        assert!(
            s.contains("- `mem-a-1` (fact, dom-demo): Retry backoff. Backoff doubles"),
            "{s}"
        );
        assert!(
            s.contains(
                "- `mem-a-2` (fact, dom-demo): Context header (retrieved with `mem-a-1` 4 times)"
            ),
            "{s}"
        );
    }

    #[test]
    fn tail_keeps_the_end() {
        assert_eq!(tail_of("  ", 10), "(nothing)");
        assert_eq!(tail_of("short", 10), "short");
        let t = tail_of("abcdefghij", 4);
        assert!(t.ends_with("ghij"), "{t}");
        assert!(t.starts_with('…'));
    }

    #[test]
    fn muse_s_answer_is_the_terminal_record_s_text() {
        let stream = concat!(
            r#"{"payload_type":"turn.input.user","payload":{"prompt":"hi"}}"#,
            "\n",
            r#"{"payload_type":"run.output.delta","payload":{"text":"gate"}}"#,
            "\n",
            r#"{"payload_type":"run.terminal.completed","payload":{"text":"gateway"}}"#,
            "\n",
        );
        assert_eq!(muse_answer(stream).as_deref(), Some("gateway"));
        assert!(muse_answer("plain text\n").is_none());
        let failed =
            r#"{"payload_type":"run.terminal.failed","payload":{"reason":"API error 400"}}"#;
        assert_eq!(
            muse_answer(failed).as_deref(),
            Some("muse: the run failed: API error 400")
        );
    }
}