supercode_harness/

mail_transcript.rs

//! What a session's transcript says about its mail: the lines its own user
//! typed, and when each message reached the model.
//!
//! The mailbox records a message as SENT: filed for its session and handed to
//! the session's door. Only the harness's transcript records it as DELIVERED:
//! in the model's context. The two differ whenever the session is busy, since
//! a message handed to a harness mid-turn waits in its queue until the running
//! tool call returns. This module reads the transcript for both halves of that.
//!
//! A line a person types into a session reaches it through the harness's own
//! composer, not through supercode, so no envelope is filed for it. Claude Code
//! records each one:
//!
//! - typed at an idle prompt: a `user` entry marked `promptSource: typed`;
//! - typed while the session was busy: a `queue-operation` `enqueue` (sent),
//!   then either a `dequeue` and a `user` entry marked `promptSource: queued`,
//!   or a `remove` (`absorbed_mid_turn`) and a `queued_command` attachment from
//!   a human (delivered).
//!
//! [`file_typed_lines`] files each as a `user` envelope in the session's
//! mailbox, so it has an id like every other message. A line waiting in the
//! queue is filed as sent; the queue also holds scheduled prompts and other
//! sessions' messages, so a queued line that lands as someone else's is taken
//! back out of the mailbox, and one removed from the queue unread is withdrawn. The id is derived from
//! the session's address, the moment the line was sent and its words, so it is
//! the same before and after the line reaches the model, and every reader
//! derives the same one. A line supercode itself typed into the session's pane
//! (a Room member's or a voice turn) is filed already under its own id.
//!
//! Mail supercode filed reaches the model as its rendering,
//! `<cross-session-message id="m-…"`, inside a prompt or a queued-command
//! attachment; a Room or voice turn reaches it as a typed line with its words.
//!
//! Codex rollouts do not record who wrote a user message (a typed line and the
//! injected AGENTS.md or environment context look alike), so their typed lines
//! are not filed; their mail is delivered once its rendering is in a user
//! message or a tool's output. Other harnesses' records are not read here:
//! their mail's delivery is unknown.

use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::io::BufRead;
use std::path::{Path, PathBuf};

use serde::Serialize;

use crate::mailbox::{Envelope, MailAddress, MailKind, Mailbox, ReplyVia, StoredEnvelope};
use crate::HarnessHomes;

/// Prefix of the id of a line a session's user typed.
pub const TYPED_ID_PREFIX: &str = "u-";

/// Prefix of the id of a channel line a harness received on its own channel
/// (Claude Code's `[channel: …]` prompts, from a Room or another bridge).
pub const CHANNEL_ID_PREFIX: &str = "c-";

/// One line a session received on a channel of its harness's own: its header
/// (`[channel: <room> · from: <who> · at: <time> …]`) and words.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChannelLine {
    /// The line's header, which names its channel, sender and send time.
    pub header: String,
    /// Who sent it, as the header names them.
    pub from: String,
    /// When it was sent, in epoch milliseconds.
    pub sent_at_ms: u64,
    /// When it reached the model; `None` while it waits in the harness's queue.
    pub delivered_at_ms: Option<u64>,
    /// The line, header included.
    pub text: String,
}

/// The id of the channel line with `header` in the session at `address`.
pub fn channel_line_id(address: &MailAddress, header: &str) -> String {
    let hash = blake3::hash(format!("{address}\n{header}").as_bytes()).to_hex();
    format!("{CHANNEL_ID_PREFIX}{}", &hash[..24])
}

/// The channel lines in a prompt (several arrive as one when they waited
/// together), each from its `[channel: …]` header to the next.
fn channel_lines(text: &str, fallback_ms: u64, delivered_at_ms: Option<u64>) -> Vec<ChannelLine> {
    let starts: Vec<usize> = text
        .match_indices("[channel: ")
        .map(|(at, _)| at)
        .filter(|at| *at == 0 || text[..*at].ends_with('\n'))
        .collect();
    starts
        .iter()
        .enumerate()
        .filter_map(|(n, start)| {
            let end = starts.get(n + 1).copied().unwrap_or(text.len());
            let line = text[*start..end].trim_end();
            let header = &line[..line.find(']')? + 1];
            let field = |name: &str| {
                header
                    .split(" · ")
                    .find_map(|part| part.strip_prefix(name))
                    .map(|value| value.trim_end_matches(']').trim().to_string())
            };
            Some(ChannelLine {
                header: header.to_string(),
                from: field("from: ").unwrap_or_else(|| "a channel".to_string()),
                sent_at_ms: field("at: ")
                    .and_then(|at| supercode_interchange::sidecar::rfc3339_to_ms(&at))
                    .and_then(|at| u64::try_from(at).ok())
                    .unwrap_or(fallback_ms),
                delivered_at_ms,
                text: line.to_string(),
            })
        })
        .collect()
}

/// Prefix of the id of a session's answer to its user.
pub const ANSWER_ID_PREFIX: &str = "a-";

/// The id of the answer the transcript entry `native_id` records in the
/// session at `address`: `a-` and 24 hex digits.
pub fn answer_id(address: &MailAddress, native_id: &str) -> String {
    let hash = blake3::hash(format!("{address}\n{native_id}").as_bytes()).to_hex();
    format!("{ANSWER_ID_PREFIX}{}", &hash[..24])
}

/// The mailbox of the person who uses the sessions on `machine`: where the
/// lines they type come from and where the sessions' answers go.
pub fn user_address(machine: &str) -> std::io::Result<MailAddress> {
    MailAddress::new(machine.to_string(), "operator", "user")
        .map_err(|error| std::io::Error::other(error.0))
}

/// The id of the line with `text` that the user of the session at `address`
/// sent at `sent_at_ms`: `u-` and 24 hex digits.
pub fn typed_line_id(address: &MailAddress, sent_at_ms: u64, text: &str) -> String {
    let hash = blake3::hash(format!("{address}\n{sent_at_ms}\n{text}").as_bytes()).to_hex();
    format!("{TYPED_ID_PREFIX}{}", &hash[..24])
}

/// One line the session's user typed, as its transcript records it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TypedLine {
    /// When the user sent it, in epoch milliseconds.
    pub sent_at_ms: u64,
    /// When it reached the model; `None` while it waits in the harness's queue.
    pub delivered_at_ms: Option<u64>,
    /// Taken back out of the queue before it reached the model.
    pub withdrawn: bool,
    /// The words, as typed.
    pub text: String,
}

/// One answer the session gave its user: a turn's final message.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Answer {
    /// The transcript entry's own id.
    pub native_id: String,
    /// When it was written, in epoch milliseconds.
    pub at_ms: u64,
    /// Its words.
    pub text: String,
}

/// What one transcript says about a session's mail.
#[derive(Debug, Clone, Default)]
pub struct TranscriptMail {
    /// The lines the session's user typed, in the order they were sent.
    pub typed: Vec<TypedLine>,
    /// The session's answers to its user, in order.
    pub answers: Vec<Answer>,
    /// Lines the session received on its harness's own channels, in order.
    pub channel: Vec<ChannelLine>,
    /// Each filed message the model has seen, by id: when it first did.
    pub delivered: HashMap<String, u64>,
}

/// Where a message stands.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum Delivery {
    /// In the model's context since this moment (epoch milliseconds).
    Delivered {
        /// When it first was.
        at_ms: u64,
    },
    /// Sent to the session, not yet in its model's context.
    Sent,
    /// Taken back out of the harness's queue before it reached the model.
    Withdrawn,
    /// The session's transcript cannot be read here, so whether it reached
    /// the model is not known.
    Unknown,
    /// Its receiver's transcript was not read for this listing (it reads only
    /// the newest receivers'); `supercode message show <id>` reads it.
    NotRead,
}

impl Delivery {
    /// How an inbox listing says it.
    pub fn describe(&self) -> String {
        match self {
            Self::Delivered { at_ms } => {
                supercode_interchange::sidecar::ms_to_rfc3339(*at_ms as i64)
            }
            Self::Sent => "not yet".to_string(),
            Self::Withdrawn => "never (taken back)".to_string(),
            Self::Unknown => "unknown".to_string(),
            Self::NotRead => "not read here (supercode message show reads it)".to_string(),
        }
    }
}

/// The transcript of the session at `address` on this machine, when its
/// harness's records are read here.
pub fn transcript(homes: &HarnessHomes, address: &MailAddress) -> Option<PathBuf> {
    if address.harness != "claude-code" {
        return None;
    }
    let file = format!("{}.jsonl", address.session_id);
    std::fs::read_dir(&homes.claude_code)
        .ok()?
        .flatten()
        .map(|project| project.path().join(&file))
        .find(|path| path.is_file())
}

fn at_ms(value: &serde_json::Value) -> u64 {
    value
        .as_str()
        .and_then(supercode_interchange::sidecar::rfc3339_to_ms)
        .and_then(|at| u64::try_from(at).ok())
        .unwrap_or_default()
}

fn prompt_text(content: &serde_json::Value) -> Option<String> {
    let text = match content {
        serde_json::Value::String(text) => text.clone(),
        serde_json::Value::Array(parts) => parts
            .iter()
            .filter(|part| part["type"] == "text")
            .filter_map(|part| part["text"].as_str())
            .collect::<Vec<_>>()
            .join("\n"),
        _ => return None,
    };
    (!text.trim().is_empty()).then_some(text)
}

/// Mail ids rendered into `line` as the model reads them: the envelope's own
/// `cross-session-message id="m-…"` (escaped or not, as Claude's relay nests
/// it), or the relay's `(message m-…)` note.
fn rendered_mail_ids(line: &str) -> Vec<&str> {
    let mut ids = Vec::new();
    for mark in ["cross-session-message id=\\\"", "(message "] {
        for (at, _) in line.match_indices(mark) {
            let rest = &line[at + mark.len()..];
            let end = rest
                .find(|character: char| !(character.is_ascii_alphanumeric() || character == '-'))
                .unwrap_or(rest.len());
            let id = &rest[..end];
            if id.starts_with("m-") {
                ids.push(id);
            }
        }
    }
    ids
}

/// Whether a line handed to a harness's queue is machine-made on its face: supercode's mail or
/// notices, or a background task's report.
fn machine_made(text: &str) -> bool {
    text.contains("<cross-session-message")
        || text.starts_with("[Cross-session")
        || text.starts_with("[channel: ")
        || text.starts_with("<task-notification>")
}

/// Read a Claude Code transcript for its typed lines and delivered mail.
pub fn read_claude(path: &Path) -> std::io::Result<TranscriptMail> {
    let reader = std::io::BufReader::new(std::fs::File::open(path)?);
    let mut mail = TranscriptMail::default();
    // The harness's queue, mirrored in full (it is first in, first out): each
    // line's words and, when it may be the user's, its entry in `mail.typed`.
    // A queue also holds scheduled prompts and other sessions' messages, so a
    // line waiting there is the user's only provisionally, until it lands as
    // theirs.
    let mut queued: VecDeque<(String, Option<usize>)> = VecDeque::new();
    // How many lines the queue has let go since the last prompt: they land
    // together as the next one.
    let mut dequeued = 0_usize;
    // Lines taken into a running turn, waiting for the attachment that says
    // who wrote them.
    let mut absorbed: VecDeque<(String, Option<usize>)> = VecDeque::new();
    let take = |waiting: &mut VecDeque<(String, Option<usize>)>, text: &str| {
        let at = waiting.iter().position(|(words, _)| words == text)?;
        waiting.remove(at).map(|(_, index)| index)
    };
    // Queued lines that landed as someone else's.
    let mut not_typed: HashSet<usize> = HashSet::new();
    // When each line left the queue for a running turn. Its queued-command
    // attachment carries the time it was queued, not this one.
    let mut taken_in: HashMap<String, VecDeque<u64>> = HashMap::new();
    // Attachments written before their line's remove (older harness builds
    // write them first): their words, the mail they render, the typed line
    // they carry, and their own stamp should no remove come.
    let mut early: Vec<(String, Vec<String>, Option<usize>, u64)> = Vec::new();
    for line in reader.lines() {
        let line = line?;
        let prompt = line.contains("\"promptSource\"") || line.contains("\"type\":\"user\"");
        let queue = line.contains("\"queue-operation\"");
        let attachment = line.contains("\"queued_command\"");
        let answer = line.contains("\"stop_reason\":\"end_turn\"")
            || line.contains("\"stop_reason\":\"stop_sequence\"");
        let rendered = line.contains("cross-session-message id=") || line.contains("(message m-");
        if !(prompt || queue || attachment || rendered || answer) {
            continue;
        }
        let Ok(record) = serde_json::from_str::<serde_json::Value>(&line) else {
            continue;
        };
        if record["isSidechain"] == true {
            continue;
        }
        let kind = record["type"].as_str().unwrap_or_default();
        let mut at = at_ms(&record["timestamp"]);
        if kind == "queue-operation" && record["operation"] == "remove" {
            if let Some(text) = record["content"].as_str() {
                // The remove an early attachment waited for: it entered the
                // context now.
                if let Some(found) = early.iter().position(|(words, ..)| words == text) {
                    let (_, ids, index, _) = early.remove(found);
                    for id in ids {
                        mail.delivered.entry(id).or_insert(at);
                    }
                    if let Some(index) = index {
                        mail.typed[index].delivered_at_ms = Some(at);
                    }
                    continue;
                }
                taken_in.entry(text.to_string()).or_default().push_back(at);
            }
        }
        let mut waits_for_remove = false;
        if kind == "attachment" && record["attachment"]["type"] == "queued_command" {
            match record["attachment"]["prompt"]
                .as_str()
                .and_then(|text| taken_in.get_mut(text))
                .and_then(VecDeque::pop_front)
            {
                Some(taken) => at = taken,
                None => waits_for_remove = true,
            }
        }
        if waits_for_remove {
            let attached = &record["attachment"];
            let text = attached["prompt"].as_str().unwrap_or_default().to_string();
            let human = attached["origin"]["kind"] == "human" || attached["humanTurn"] == true;
            let ids = rendered_mail_ids(&line)
                .into_iter()
                .map(str::to_string)
                .collect();
            // Its line leaves the queue with the remove still to come.
            let index = match take(&mut queued, &text) {
                Some(Some(index)) if !human => {
                    not_typed.insert(index);
                    None
                }
                Some(index) => index,
                None if human && !text.trim().is_empty() => {
                    mail.typed.push(TypedLine {
                        sent_at_ms: at,
                        delivered_at_ms: None,
                        withdrawn: false,
                        text: text.clone(),
                    });
                    Some(mail.typed.len() - 1)
                }
                None => None,
            };
            early.push((text, ids, index, at));
            continue;
        }
        // Mail in the model's context: a prompt, an attachment or a tool's
        // result rendering it; never an assistant's own words naming an id,
        // nor the queue still holding it.
        if rendered && matches!(kind, "user" | "attachment") {
            for id in rendered_mail_ids(&line) {
                mail.delivered.entry(id.to_string()).or_insert(at);
            }
        }
        match kind {
            // A turn's final message is its answer to the user.
            "assistant"
                if answer
                    && matches!(
                        record["message"]["stop_reason"].as_str(),
                        Some("end_turn" | "stop_sequence")
                    ) =>
            {
                let (Some(native_id), Some(text)) = (
                    record["uuid"].as_str(),
                    prompt_text(&record["message"]["content"]),
                ) else {
                    continue;
                };
                mail.answers.push(Answer {
                    native_id: native_id.to_string(),
                    at_ms: at,
                    text,
                });
            }
            // A dequeue names no line: the queue lets go of its oldest.
            "queue-operation" if record["operation"] == "dequeue" => dequeued += 1,
            "queue-operation" => {
                let Some(text) = record["content"].as_str() else {
                    continue;
                };
                match record["operation"].as_str() {
                    Some("enqueue") => {
                        let index = (!machine_made(text) && !text.trim().is_empty()).then(|| {
                            mail.typed.push(TypedLine {
                                sent_at_ms: at,
                                delivered_at_ms: None,
                                withdrawn: false,
                                text: text.to_string(),
                            });
                            mail.typed.len() - 1
                        });
                        queued.push_back((text.to_string(), index));
                    }
                    Some("remove") => {
                        let Some(index) = take(&mut queued, text) else {
                            continue;
                        };
                        match record["reason"].as_str() {
                            // Taken into the running turn: its attachment follows.
                            Some("absorbed_mid_turn" | "delivered_to_agent") => {
                                if let Some(index) = index {
                                    mail.typed[index].delivered_at_ms = Some(at);
                                }
                                absorbed.push_back((text.to_string(), index));
                            }
                            // Taken back before it reached the model.
                            _ => {
                                if let Some(index) = index {
                                    mail.typed[index].withdrawn = true;
                                }
                            }
                        }
                    }
                    _ => {}
                }
            }
            // A line taken into a running turn lands as a queued command; it
            // is the user's when its origin is a person.
            "attachment" => {
                let attached = &record["attachment"];
                let Some(text) = attached["prompt"].as_str() else {
                    continue;
                };
                if attached["type"] != "queued_command" || text.trim().is_empty() {
                    continue;
                }
                let human = attached["origin"]["kind"] == "human" || attached["humanTurn"] == true;
                match take(&mut absorbed, text) {
                    Some(Some(index)) if !human => {
                        not_typed.insert(index);
                    }
                    Some(_) => {}
                    None if human => mail.typed.push(TypedLine {
                        sent_at_ms: at_ms(&attached["timestamp"]).min(at),
                        delivered_at_ms: Some(at),
                        withdrawn: false,
                        text: text.to_string(),
                    }),
                    None => {}
                }
            }
            "user" => {
                let Some(text) = prompt_text(&record["message"]["content"]) else {
                    continue;
                };
                let source = record["promptSource"].as_str();
                let human = matches!(source, Some("typed" | "queued")) && record["isMeta"] != true;
                if !human {
                    mail.channel.extend(channel_lines(&text, at, Some(at)));
                }
                // Queued lines land as the next prompt when the turn they waited
                // for ends, several at once when several were let go: the lines
                // whose words it carries (the queue's oldest when it carries
                // none, since the transcript does not record every way a queue
                // empties, and matching by words keeps one lost line from
                // shifting every later one).
                let mut landing: Vec<Option<usize>> = Vec::new();
                let let_go = std::mem::take(&mut dequeued);
                if let_go == 0 {
                    landing.extend(take(&mut queued, &text));
                }
                let wanted = let_go;
                while landing.len() < wanted {
                    let Some(at) = queued
                        .iter()
                        .position(|(words, _)| text.contains(words.as_str()))
                    else {
                        break;
                    };
                    landing.extend(queued.remove(at).map(|(_, index)| index));
                }
                if landing.is_empty() && wanted > 0 {
                    for _ in 0..wanted {
                        landing.extend(queued.pop_front().map(|(_, index)| index));
                    }
                }
                if !landing.is_empty() {
                    for index in landing.into_iter().flatten() {
                        if human {
                            mail.typed[index].delivered_at_ms = Some(at);
                        } else {
                            not_typed.insert(index);
                        }
                    }
                    continue;
                }
                if human {
                    mail.typed.push(TypedLine {
                        sent_at_ms: at,
                        delivered_at_ms: Some(at),
                        withdrawn: false,
                        text,
                    });
                }
            }
            _ => {}
        }
    }
    // An early attachment whose remove never came is in the context all the
    // same: it counts from its own stamp.
    for (_, ids, index, at) in early {
        for id in ids {
            mail.delivered.entry(id).or_insert(at);
        }
        if let Some(index) = index {
            mail.typed[index].delivered_at_ms.get_or_insert(at);
        }
    }
    // A line still queued stays provisional: sent, not delivered.
    for (text, _) in &queued {
        if text.starts_with("[channel: ") {
            mail.channel.extend(channel_lines(text, 0, None));
        }
    }
    let mut index = 0;
    mail.typed.retain(|_| {
        index += 1;
        !not_typed.contains(&(index - 1))
    });
    Ok(mail)
}

/// The transcript's reading of the session at `address`, when it can be read.
pub fn read(homes: &HarnessHomes, address: &MailAddress) -> Option<TranscriptMail> {
    if address.harness == "codex" {
        return read_codex_answers(&crate::mail_question::transcript_for(address)?).ok();
    }
    read_claude(&transcript(homes, address)?).ok()
}

/// Final answers and delivered mail from a native Codex pane's own transcript.
/// Question replies have separate native receipts; neither tool outputs nor
/// developer context are invented as typed user turns here.
fn read_codex_answers(path: &Path) -> std::io::Result<TranscriptMail> {
    let reader = std::io::BufReader::new(std::fs::File::open(path)?);
    let mut mail = TranscriptMail::default();
    for line in reader.lines().map_while(Result::ok) {
        let rendered = line.contains("cross-session-message id=");
        if !rendered && !line.contains("\"final_answer\"") {
            continue;
        }
        let Ok(record) = serde_json::from_str::<serde_json::Value>(&line) else {
            continue;
        };
        let item = &record["payload"];
        // Mail in the model's context, as Claude's reading counts it: a user
        // message (a pasted turn) or a tool's output (`supercode message inbox`
        // run in a turn) rendering it; never the model's own words naming an id,
        // nor the event stream echoing those items.
        if rendered
            && record["type"] == "response_item"
            && ((item["type"] == "message" && item["role"] == "user")
                || matches!(
                    item["type"].as_str(),
                    Some("function_call_output" | "custom_tool_call_output")
                ))
        {
            let at = at_ms(&record["timestamp"]);
            for id in codex_rendered_mail_ids(&line) {
                mail.delivered.entry(id.to_string()).or_insert(at);
            }
        }
        if record["type"] != "response_item"
            || item["type"] != "message"
            || item["role"] != "assistant"
            || item["phase"] != "final_answer"
        {
            continue;
        }
        let Some(id) = item["id"].as_str() else {
            continue;
        };
        let text = item["content"]
            .as_array()
            .into_iter()
            .flatten()
            .filter(|part| part["type"] == "output_text")
            .filter_map(|part| part["text"].as_str())
            .collect::<Vec<_>>()
            .join("\n");
        if !text.is_empty() {
            mail.answers.push(Answer {
                native_id: id.to_string(),
                at_ms: at_ms(&record["timestamp"]),
                text,
            });
        }
    }
    Ok(mail)
}

/// Mail ids rendered into a Codex rollout line: `cross-session-message id="m-…"`
/// with the quote escaped once (a user message) or more (a tool's output is
/// JSON text inside the record's JSON).
fn codex_rendered_mail_ids(line: &str) -> Vec<&str> {
    let mark = "cross-session-message id=";
    line.match_indices(mark)
        .filter_map(|(at, _)| {
            let rest = line[at + mark.len()..].trim_start_matches(['\\', '"']);
            let end = rest
                .find(|character: char| !(character.is_ascii_alphanumeric() || character == '-'))
                .unwrap_or(rest.len());
            Some(&rest[..end]).filter(|id| id.starts_with("m-"))
        })
        .collect()
}

/// The newest of `items` (time, id), sorted by time, sent before `at` (or at
/// it, when `inclusive`).
fn newest_before(items: &[(u64, String)], at: u64, inclusive: bool) -> Option<String> {
    items
        .iter()
        .rev()
        .find(|(when, _)| if inclusive { *when <= at } else { *when < at })
        .map(|(_, id)| id.clone())
}

/// Each answer the session gave, as (when, id), oldest first.
fn answer_times(address: &MailAddress, mail: &TranscriptMail) -> Vec<(u64, String)> {
    let mut times: Vec<(u64, String)> = mail
        .answers
        .iter()
        .map(|answer| (answer.at_ms, answer_id(address, &answer.native_id)))
        .collect();
    times.sort();
    times
}

/// File `envelope` unless it is filed as it stands; a record an earlier
/// reading filed differently (without its inferred reply) is filed again.
fn file_derived(
    mailbox: &Mailbox,
    filed: &HashMap<&str, &StoredEnvelope>,
    envelope: Envelope,
) -> std::io::Result<bool> {
    if let Some(stored) = filed.get(envelope.id.as_str()) {
        if stored.envelope.in_reply_to == envelope.in_reply_to {
            return Ok(false);
        }
        std::fs::remove_file(&stored.path).ok();
    }
    mailbox.file_read(&envelope)?;
    Ok(true)
}

/// File every line the user of `mailbox`'s session typed that is not filed
/// yet, reading the transcript as `mail`. Returns how many were filed now.
pub fn file_typed_lines(mailbox: &Mailbox, mail: &TranscriptMail) -> std::io::Result<usize> {
    let address = mailbox.address();
    let filed = mailbox.list()?;
    let by_id: HashMap<&str, &StoredEnvelope> = filed
        .iter()
        .map(|stored| (stored.envelope.id.as_str(), stored))
        .collect();
    // A line a person sends answers the session's newest answer before it:
    // the reply is inferred, as a mail client threads a reply by its subject.
    let answers = answer_times(address, mail);
    // What supercode typed into the pane is filed already, under its own id:
    // each such turn accounts for one typed line with its words.
    let mut typed_by_supercode: HashMap<&str, usize> = HashMap::new();
    for stored in &filed {
        if stored.envelope.kind == MailKind::User
            && !stored.envelope.id.starts_with(TYPED_ID_PREFIX)
        {
            *typed_by_supercode
                .entry(stored.envelope.body.trim())
                .or_default() += 1;
        }
    }
    let user = user_address(&address.machine)?;
    // A typed line's record is the transcript's reading: one the reading no
    // longer yields (a queued line that landed as a scheduled prompt's or a
    // session's, or one an older reading took for the user's) is taken back.
    let current: HashSet<String> = mail
        .typed
        .iter()
        .map(|line| typed_line_id(address, line.sent_at_ms, &line.text))
        .collect();
    // A Codex transcript does not say who wrote a prompt: its typed lines come from its
    // UserPromptSubmit hook ([`file_prompt`]), never from this reading, so none is taken back here.
    let prunes = address.harness != "codex";
    for stored in &filed {
        // Only this session's own records (user mail): a line typed in another session and copied here as CC
        // (`MailKind::Typed`, docs/adr/0008) shares the id prefix but is never this transcript's to take back.
        if prunes
            && stored.envelope.kind == MailKind::User
            && stored.envelope.id.starts_with(TYPED_ID_PREFIX)
            && !stored
                .envelope
                .in_reply_to
                .as_deref()
                .is_some_and(|id| id.starts_with("q-"))
            && !current.contains(&stored.envelope.id)
        {
            std::fs::remove_file(&stored.path).ok();
        }
    }
    let mut count = 0;
    for line in &mail.typed {
        let id = typed_line_id(address, line.sent_at_ms, &line.text);
        if !by_id.contains_key(id.as_str()) {
            if let Some(left) = typed_by_supercode.get_mut(line.text.trim()) {
                if *left > 0 {
                    *left -= 1;
                    continue;
                }
            }
        }
        let in_reply_to = newest_before(&answers, line.sent_at_ms, false);
        count += usize::from(file_derived(
            mailbox,
            &by_id,
            Envelope {
                id,
                created_at_ms: line.sent_at_ms,
                from: user.clone(),
                from_name: format!("user@{}", address.machine),
                sender_identity: None,
                kind: MailKind::User,
                reply_via: ReplyVia::None,
                in_reply_to_inferred: in_reply_to.is_some(),
                in_reply_to,
                thread: None,
                native_from: None,
                voice_for: None,
                subject: None,
                body: line.text.clone(),
            },
        )?);
    }
    for line in &mail.channel {
        let id = channel_line_id(address, &line.header);
        let in_reply_to = newest_before(&answers, line.sent_at_ms, false);
        count += usize::from(file_derived(
            mailbox,
            &by_id,
            Envelope {
                id,
                created_at_ms: line.sent_at_ms,
                from: MailAddress::new(address.machine.clone(), "operator", "channel")
                    .map_err(|error| std::io::Error::other(error.0))?,
                from_name: line.from.clone(),
                sender_identity: None,
                kind: MailKind::Channel,
                reply_via: ReplyVia::None,
                in_reply_to_inferred: in_reply_to.is_some(),
                in_reply_to,
                thread: None,
                native_from: None,
                voice_for: None,
                subject: None,
                body: line.text.clone(),
            },
        )?);
    }
    Ok(count)
}

/// File a line a person typed into the session at `address`, as its harness's
/// UserPromptSubmit hook reports it, for a harness whose transcript does not say
/// who wrote a prompt (Codex). Supercode's own mail (its envelopes, and the user's
/// turns it typed, filed under their own ids) is not a typed line. Returns whether
/// it was filed now.
pub fn file_prompt(address: &MailAddress, text: &str, sent_at_ms: u64) -> std::io::Result<bool> {
    if text.trim().is_empty() || machine_made(text) {
        return Ok(false);
    }
    let mailbox = Mailbox::open(&crate::mailbox::mail_root(), address)?;
    let filed = mailbox.list()?;
    let id = typed_line_id(address, sent_at_ms, text);
    let typed_by_supercode = filed.iter().any(|stored| {
        stored.envelope.kind == MailKind::User
            && !stored.envelope.id.starts_with(TYPED_ID_PREFIX)
            && stored.envelope.body.trim() == text.trim()
            && sent_at_ms.saturating_sub(stored.envelope.created_at_ms) < 15 * 60 * 1000
    });
    if typed_by_supercode || filed.iter().any(|stored| stored.envelope.id == id) {
        return Ok(false);
    }
    mailbox.file_read(&Envelope {
        id,
        created_at_ms: sent_at_ms,
        from: user_address(&address.machine)?,
        from_name: format!("user@{}", address.machine),
        sender_identity: None,
        kind: MailKind::User,
        reply_via: ReplyVia::None,
        in_reply_to: None,
        in_reply_to_inferred: false,
        thread: None,
        native_from: None,
        voice_for: None,
        subject: None,
        body: text.to_string(),
    })?;
    Ok(true)
}

/// File every answer the session at `address` gave its user that is not filed
/// yet, in the user's mailbox under `root`, from the session (`name` is what
/// the user knows it by). Returns how many were filed now.
pub fn file_answers(
    root: &Path,
    address: &MailAddress,
    name: &str,
    mail: &TranscriptMail,
) -> std::io::Result<usize> {
    let recipient =
        crate::mail_question::creator_for(address).unwrap_or(user_address(&address.machine)?);
    let mailbox = Mailbox::open(root, &recipient)?;
    let listed = mailbox.list()?;
    let mut filed: HashMap<&str, &StoredEnvelope> = HashMap::new();
    for stored in &listed {
        // An answer an earlier reading filed as a session's mail is filed again as an answer.
        if stored.envelope.id.starts_with(ANSWER_ID_PREFIX)
            && stored.envelope.kind != MailKind::Answer
        {
            std::fs::remove_file(&stored.path).ok();
            continue;
        }
        filed.insert(stored.envelope.id.as_str(), stored);
    }
    // An answer answers the newest line a person sent the session before it: a
    // line typed or spoken to it, a Room member's turn, a channel line.
    let mut people: Vec<(u64, String)> = mail
        .typed
        .iter()
        .map(|line| {
            (
                line.sent_at_ms,
                typed_line_id(address, line.sent_at_ms, &line.text),
            )
        })
        .chain(
            mail.channel
                .iter()
                .map(|line| (line.sent_at_ms, channel_line_id(address, &line.header))),
        )
        .chain(
            Mailbox::open(root, address)?
                .list()?
                .into_iter()
                .filter(|stored| {
                    stored.envelope.kind == MailKind::User
                        && (!stored.envelope.id.starts_with(TYPED_ID_PREFIX)
                            || stored
                                .envelope
                                .in_reply_to
                                .as_deref()
                                .is_some_and(|id| id.starts_with("q-")))
                })
                .map(|stored| (stored.envelope.created_at_ms, stored.envelope.id)),
        )
        .collect();
    people.sort();
    let mut count = 0;
    for answer in &mail.answers {
        let id = answer_id(address, &answer.native_id);
        let in_reply_to = newest_before(&people, answer.at_ms, true);
        let envelope = Envelope {
            id,
            created_at_ms: answer.at_ms,
            from: address.clone(),
            from_name: name.to_string(),
            sender_identity: None,
            kind: MailKind::Answer,
            reply_via: ReplyVia::None,
            in_reply_to_inferred: in_reply_to.is_some(),
            in_reply_to,
            thread: None,
            native_from: None,
            voice_for: None,
            subject: None,
            body: answer.text.clone(),
        };
        if recipient.machine != address.machine && !filed.contains_key(envelope.id.as_str()) {
            // Keep the local outbound mirror only after the creator's machine
            // accepts it. A failed remote delivery can be retried on the next read.
            crate::mailbox::deliver_to(&recipient, &envelope)?;
        }
        count += usize::from(file_derived(&mailbox, &filed, envelope)?);
    }
    Ok(count)
}

/// Everything the session at `address` sent that is filed on this machine
/// under `root`: its messages to other sessions and its answers to its user,
/// each with the mailbox it is filed in, oldest first.
pub fn sent_by(root: &Path, address: &MailAddress) -> Vec<(MailAddress, StoredEnvelope)> {
    let mut sent: Vec<(MailAddress, StoredEnvelope)> = crate::mailbox::all_mailboxes(root)
        .into_iter()
        .filter(|mailbox| mailbox.address() != address)
        .flat_map(|mailbox| {
            let to = mailbox.address().clone();
            mailbox
                .list()
                .unwrap_or_default()
                .into_iter()
                .filter(|stored| &stored.envelope.from == address)
                .map(move |stored| (to.clone(), stored))
        })
        .collect();
    sent.sort_by_key(|(_, stored)| stored.envelope.created_at_ms);
    sent
}

/// Where each of `stored` (the mailbox of the session at `address`) stands,
/// by id, as the transcript reading `mail` shows it (`None`: unreadable).
pub fn deliveries(
    address: &MailAddress,
    stored: &[StoredEnvelope],
    mail: Option<&TranscriptMail>,
) -> BTreeMap<String, Delivery> {
    let Some(mail) = mail else {
        return stored
            .iter()
            .map(|stored| {
                (
                    stored.envelope.id.clone(),
                    crate::mail_question::delivered_at(&stored.envelope)
                        .map(|at_ms| Delivery::Delivered { at_ms })
                        .unwrap_or(Delivery::Unknown),
                )
            })
            .collect();
    };
    let typed: HashMap<String, &TypedLine> = mail
        .typed
        .iter()
        .map(|line| (typed_line_id(address, line.sent_at_ms, &line.text), line))
        .collect();
    let channel: HashMap<String, &ChannelLine> = mail
        .channel
        .iter()
        .map(|line| (channel_line_id(address, &line.header), line))
        .collect();
    // A turn supercode typed into the pane reaches the model as a typed line
    // with its words, the first one sent after it was filed.
    let mut used: HashSet<usize> = HashSet::new();
    let mut result = BTreeMap::new();
    for stored in stored {
        let envelope = &stored.envelope;
        if typed.get(&envelope.id).is_some_and(|line| line.withdrawn) {
            result.insert(envelope.id.clone(), Delivery::Withdrawn);
            continue;
        }
        let at = if let Some(line) = typed.get(&envelope.id) {
            line.delivered_at_ms
        } else if let Some(line) = channel.get(&envelope.id) {
            line.delivered_at_ms
        } else if let Some(at) = mail
            .delivered
            .get(&envelope.id)
            .or_else(|| mail.delivered.get(crate::mailbox::short_id(&envelope.id)))
        {
            Some(*at)
        } else if envelope.kind == MailKind::User
            && envelope
                .in_reply_to
                .as_deref()
                .is_some_and(|id| id.starts_with("q-"))
        {
            crate::mail_question::delivered_at(envelope)
        } else if envelope.kind == MailKind::User {
            mail.typed
                .iter()
                .enumerate()
                .find(|(index, line)| {
                    !used.contains(index)
                        && line.sent_at_ms + 1_000 >= envelope.created_at_ms
                        && line.text.trim() == envelope.body.trim()
                })
                .and_then(|(index, line)| {
                    used.insert(index);
                    line.delivered_at_ms
                })
        } else {
            None
        };
        result.insert(
            envelope.id.clone(),
            match at {
                Some(at_ms) => Delivery::Delivered { at_ms },
                None => Delivery::Sent,
            },
        );
    }
    result
}

/// Fewest hex digits after the kind (`m-`, `u-`, `a-`) a prefix may give.
pub const MIN_PREFIX_DIGITS: usize = 4;

/// The messages filed on this machine under `root` whose id is `id` or
/// starts with it, like git's short hashes: the mailbox holding each, and the
/// envelope. A prefix shorter than the kind and [`MIN_PREFIX_DIGITS`] digits
/// matches nothing; more than one match means the prefix is ambiguous.
pub fn find_messages(root: &Path, id: &str) -> std::io::Result<Vec<(Mailbox, StoredEnvelope)>> {
    Ok(find_messages_many(root, &[id])?.remove(0))
}

/// [`find_messages`] for several ids in one pass over this machine's mailboxes: one list per id,
/// in the order given.
pub fn find_messages_many(
    root: &Path,
    ids: &[&str],
) -> std::io::Result<Vec<Vec<(Mailbox, StoredEnvelope)>>> {
    let mut found: Vec<Vec<(Mailbox, StoredEnvelope)>> = ids.iter().map(|_| Vec::new()).collect();
    // A prefix shorter than the kind and MIN_PREFIX_DIGITS digits matches nothing.
    let usable: Vec<usize> = (0..ids.len())
        .filter(|&index| {
            ids[index]
                .split_once('-')
                .map_or(0, |(_, digits)| digits.len())
                >= MIN_PREFIX_DIGITS
        })
        .collect();
    if usable.is_empty() {
        return Ok(found);
    }
    let prefixes: Vec<&str> = usable.iter().map(|&index| ids[index]).collect();
    for mailbox in crate::mailbox::all_mailboxes(root) {
        for (matched, stored) in mailbox.find_prefixes(&prefixes)? {
            let list = &mut found[usable[matched]];
            if !list
                .iter()
                .any(|(_, known)| known.envelope.id == stored.envelope.id)
            {
                list.push((mailbox.clone(), stored));
            }
        }
    }
    Ok(found)
}