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. 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), and other harnesses'
//! records are not read here: their typed lines are not filed and 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 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.
    pub delivered_at_ms: Option<u64>,
    /// 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>,
    /// 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,
    /// The session's transcript cannot be read here, so whether it reached
    /// the model is not known.
    Unknown,
}

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::Unknown => "unknown".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
}

/// 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();
    // Lines handed to the harness while the session was busy, waiting in its
    // queue: their words and when they were sent. A queue holds notices and
    // other sessions' messages too, so a line counts as the user's only when
    // it lands as theirs.
    let mut queued: VecDeque<(String, u64)> = VecDeque::new();
    // Lines taken into a running turn: their words and when.
    let mut absorbed: VecDeque<(String, u64)> = VecDeque::new();
    let take = |waiting: &mut VecDeque<(String, u64)>, text: &str| {
        let at = waiting.iter().position(|(words, _)| words == text)?;
        waiting.remove(at).map(|(_, when)| when)
    };
    for line in reader.lines() {
        let line = line?;
        let prompt = line.contains("\"promptSource\"");
        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 at = at_ms(&record["timestamp"]);
        // 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,
                });
            }
            "queue-operation" => {
                let Some(text) = record["content"].as_str() else {
                    continue;
                };
                match record["operation"].as_str() {
                    Some("enqueue") => queued.push_back((text.to_string(), at)),
                    Some("remove") => {
                        if take(&mut queued, text).is_some() {
                            absorbed.push_back((text.to_string(), at));
                        }
                    }
                    _ => {}
                }
            }
            // 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 human = attached["origin"]["kind"] == "human" || attached["humanTurn"] == true;
                let Some(text) = attached["prompt"].as_str() else {
                    continue;
                };
                if attached["type"] != "queued_command" || !human || text.trim().is_empty() {
                    continue;
                }
                let delivered = take(&mut absorbed, text).unwrap_or(at);
                mail.typed.push(TypedLine {
                    sent_at_ms: at_ms(&attached["timestamp"]).min(delivered),
                    delivered_at_ms: Some(delivered),
                    text: text.to_string(),
                });
            }
            "user" => {
                let source = record["promptSource"].as_str();
                if !matches!(source, Some("typed" | "queued")) || record["isMeta"] == true {
                    continue;
                }
                let Some(text) = prompt_text(&record["message"]["content"]) else {
                    continue;
                };
                let sent = match source {
                    Some("queued") => take(&mut queued, &text).unwrap_or(at),
                    _ => at,
                };
                mail.typed.push(TypedLine {
                    sent_at_ms: sent,
                    delivered_at_ms: Some(at),
                    text,
                });
            }
            _ => {}
        }
    }
    // A line still in the queue is not filed: until it lands, the transcript
    // does not say who wrote it.
    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> {
    read_claude(&transcript(homes, address)?).ok()
}

/// 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 ids: HashSet<&str> = filed
        .iter()
        .map(|stored| stored.envelope.id.as_str())
        .collect();
    // 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)?;
    let mut count = 0;
    for line in &mail.typed {
        let id = typed_line_id(address, line.sent_at_ms, &line.text);
        if ids.contains(id.as_str()) {
            continue;
        }
        if let Some(left) = typed_by_supercode.get_mut(line.text.trim()) {
            if *left > 0 {
                *left -= 1;
                continue;
            }
        }
        mailbox.file_read(&Envelope {
            id,
            created_at_ms: line.sent_at_ms,
            from: user.clone(),
            from_name: format!("user@{}", address.machine),
            kind: MailKind::User,
            reply_via: ReplyVia::None,
            in_reply_to: None,
            in_reply_to_inferred: false,
            native_from: None,
            body: line.text.clone(),
        })?;
        count += 1;
    }
    Ok(count)
}

/// 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 mailbox = Mailbox::open(root, &user_address(&address.machine)?)?;
    let filed: HashSet<String> = mailbox
        .list()?
        .into_iter()
        .map(|stored| stored.envelope.id)
        .collect();
    let mut count = 0;
    for answer in &mail.answers {
        let id = answer_id(address, &answer.native_id);
        if filed.contains(&id) {
            continue;
        }
        mailbox.file_read(&Envelope {
            id,
            created_at_ms: answer.at_ms,
            from: address.clone(),
            from_name: name.to_string(),
            kind: MailKind::Peer,
            reply_via: ReplyVia::None,
            in_reply_to: None,
            in_reply_to_inferred: false,
            native_from: None,
            body: answer.text.clone(),
        })?;
        count += 1;
    }
    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(), 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();
    // 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;
        let at = if let Some(line) = typed.get(&envelope.id) {
            line.delivered_at_ms
        } else if let Some(at) = mail.delivered.get(&envelope.id) {
            Some(*at)
        } 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
}

/// Where the message `id` is filed on this machine: the mailbox of the
/// session that received it, and the envelope.
pub fn find_message(root: &Path, id: &str) -> std::io::Result<Option<(Mailbox, StoredEnvelope)>> {
    for mailbox in crate::mailbox::all_mailboxes(root) {
        if let Some(stored) = mailbox.find(id)? {
            return Ok(Some((mailbox, stored)));
        }
    }
    Ok(None)
}