clawmates/crates/cm-api/src/mission_events.rs

//! Structured mission activity — the channel that replaced parsing prose.
//!
//! The operator decision behind this module: action detail comes from
//! **structured events at the source**, never from `checkpoint.log` or model
//! output. A tool name in a log line is indistinguishable from an agent
//! *discussing* a tool, and a visualization built on that distinction reads as
//! confident fact while being partly fiction.
//!
//! Everything here is best-effort. A mission must not fail because its
//! telemetry could not be written — so every write logs and swallows. That is a
//! deliberate exception to this codebase's usual rule, and it is bounded: the
//! only thing lost is detail in a picture.

use serde_json::Value;
use sqlx::PgPool;
use uuid::Uuid;

/// A phase entered `running`.
pub const PHASE_STARTED: &str = "phase.started";
/// A phase reached a terminal state. `detail.status` says which.
pub const PHASE_COMPLETED: &str = "phase.completed";
/// An agent called a tool. `target` is the tool name.
pub const TOOL_CALL: &str = "tool.call";
/// A tool touched a path. `target` is the path, repo-relative where known.
pub const FILE_TOUCH: &str = "file.touch";
/// The exact prompt text an agent was given. `detail.text` is the full string,
/// `target` is the role or tier that composed it.
///
/// The durable answer to "what did this agent actually receive". Skills, the
/// task, the evaluator's feedback and the tool preamble are assembled from four
/// places across three tiers, so re-deriving the prompt after the fact means
/// re-running that assembly against data that has since changed. Recording it
/// is the only way the question stays answerable.
pub const PROMPT_COMPOSED: &str = "prompt.composed";
/// The agent's own narrative for a turn. `detail.text`.
///
/// Written by `topology_worker` and pushed live once by `live_bus`. Until the
/// reader below existed, the stored row was never read again by anything: both
/// database readers in `routes/world.rs` filter to `tool.call`/`file.touch`,
/// and the only other statement touching the table is the GC that deletes it.
pub const REASONING: &str = "reasoning";

/// Kinds the per-phase cap applies to.
///
/// The cap exists to bound the two unbounded kinds: a coding phase can call
/// thousands of tools and touch thousands of paths. The others are bounded by
/// the phase's own structure — one start, one completion, one prompt per turn —
/// and counting them against the same budget meant a busy phase could push out
/// its OWN terminal event, leaving a phase that looks like it never finished.
const CAPPED_KINDS: &[&str] = &[TOOL_CALL, FILE_TOUCH];

/// Does this kind count against, and get dropped by, `PER_PHASE_CAP`?
pub fn is_capped(kind: &str) -> bool {
    CAPPED_KINDS.contains(&kind)
}

/// Most events one phase may record.
///
/// A capped stream that says so beats an uncapped one that quietly becomes the
/// largest table in the database: a coding phase can call thousands of tools,
/// and every one of them would be replayed to every World subscriber. Past the
/// cap the picture is already complete — nobody reads the four-thousandth file
/// orb.
pub const PER_PHASE_CAP: i64 = 400;

/// One recorded event.
#[derive(Debug, Clone, Default)]
pub struct MissionEvent {
    pub mission_id: Uuid,
    pub phase_id: Option<Uuid>,
    pub run_id: Option<Uuid>,
    pub agent_id: Option<Uuid>,
    pub kind: String,
    pub target: Option<String>,
    pub detail: Value,
}

impl MissionEvent {
    pub fn new(mission_id: Uuid, kind: &str) -> Self {
        MissionEvent {
            mission_id,
            kind: kind.to_string(),
            detail: Value::Null,
            ..Default::default()
        }
    }
    pub fn phase(mut self, id: Uuid) -> Self {
        self.phase_id = Some(id);
        self
    }
    pub fn run(mut self, id: Uuid) -> Self {
        self.run_id = Some(id);
        self
    }
    pub fn agent(mut self, id: Option<Uuid>) -> Self {
        self.agent_id = id;
        self
    }
    pub fn target(mut self, t: impl Into<String>) -> Self {
        self.target = Some(t.into());
        self
    }
    pub fn detail(mut self, d: Value) -> Self {
        self.detail = d;
        self
    }
}

/// Record one event, best-effort.
///
/// The per-phase cap is enforced in the INSERT itself rather than by a read
/// followed by a write: two tool taps writing concurrently would both read a
/// count below the cap and both insert, and the cap would drift by however many
/// writers there are. `INSERT … SELECT … WHERE (subquery) < cap` makes the
/// decision inside the statement.
pub async fn record(pool: &PgPool, e: MissionEvent) {
    let detail = if e.detail.is_null() {
        Value::Object(Default::default())
    } else {
        e.detail
    };
    // The cap is still decided INSIDE the insert (see the test below), and now
    // only counts the kinds it is meant to bound.
    let capped = is_capped(&e.kind);
    let res = sqlx::query(
        "INSERT INTO mission_events
             (mission_id, phase_id, run_id, agent_id, kind, target, detail)
         SELECT $1, $2, $3, $4, $5, $6, $7
          WHERE $2::uuid IS NULL
             OR NOT $9
             OR (SELECT count(*) FROM mission_events
                  WHERE phase_id = $2 AND kind = ANY($10)) < $8",
    )
    .bind(e.mission_id)
    .bind(e.phase_id)
    .bind(e.run_id)
    .bind(e.agent_id)
    .bind(&e.kind)
    .bind(&e.target)
    .bind(&detail)
    .bind(PER_PHASE_CAP)
    .bind(capped)
    .bind(CAPPED_KINDS)
    .execute(pool)
    .await;
    if let Err(err) = res {
        eprintln!("mission_events: record {} failed: {err}", e.kind);
    }
}

/// Record several events under one round trip's worth of intent.
/// Every recorded prompt and narrative for a mission, oldest first.
///
/// The read side of `PROMPT_COMPOSED` / `REASONING`. Both kinds were write-only
/// before this: the prompt was never stored at all, and the narrative was
/// stored and then read by nothing. Together they answer "what did this agent
/// receive, and what did it say it did", which is the question
/// `docs/PROVENANCE-ASSESSMENT.md` records as unanswerable.
pub async fn narrative_for_mission(
    pool: &PgPool,
    mission_id: Uuid,
) -> Result<Vec<(String, Option<Uuid>, Option<String>, String)>, sqlx::Error> {
    let rows: Vec<(String, Option<Uuid>, Option<String>, Value)> = sqlx::query_as(
        "SELECT kind, agent_id, target, detail
           FROM mission_events
          WHERE mission_id = $1 AND kind = ANY($2)
          ORDER BY id",
    )
    .bind(mission_id)
    .bind(&[PROMPT_COMPOSED, REASONING][..])
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|(kind, agent, target, detail)| {
            let text = detail
                .get("text")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string();
            (kind, agent, target, text)
        })
        .collect())
}

/// One action an agent took, as a reader gets it back.
#[derive(Debug, Clone, PartialEq)]
pub struct ToolEvidence {
    /// The tool's name, e.g. `Bash`, `Write`.
    pub tool: String,
    /// The absolute path inside the sandbox, when the tool named one.
    ///
    /// Absolute, unlike the sibling `file.touch` row's `target`. See the note
    /// in `phase_runner::record_vm_tools`: normalising is what destroys the
    /// only question a path can settle.
    pub path: Option<String>,
    /// The tool's arguments, bounded by `vm_tool_tap::bounded_input`.
    pub input: Value,
    /// What a command produced, bounded by `vm_tool_tap::bounded_response`.
    ///
    /// Null for every tool that is not a command. This is where a failing test
    /// run is visible, and it is the only place it is — the recorded stream has
    /// no exit codes.
    pub response: Value,
}

impl ToolEvidence {
    /// The shell command, for the tools that run one.
    pub fn command(&self) -> Option<&str> {
        self.input.get("command").and_then(Value::as_str)
    }
}

/// Every tool call recorded for a mission, in order.
///
/// The counterpart to [`narrative_for_mission`], and the reason it exists: the
/// narrative is what an agent *said* it did. These rows are what it did. A
/// measurement built on the narrative alone scores prose, and prose is written
/// by the thing being measured.
///
/// **Bounded by [`PER_PHASE_CAP`].** A phase that ran more tools than the cap
/// returns the first `PER_PHASE_CAP` and no marker saying so, so a check that
/// concludes "this never happened" from an empty result is only sound for
/// phases under the cap. Every check in `skill_use` is one-sided in the safe
/// direction for that reason: it reports a violation it can see, never
/// compliance it inferred from silence.
pub async fn tool_evidence_for_mission(
    pool: &PgPool,
    mission_id: Uuid,
) -> Result<Vec<ToolEvidence>, sqlx::Error> {
    let rows: Vec<(Option<String>, Value)> = sqlx::query_as(
        "SELECT target, detail
           FROM mission_events
          WHERE mission_id = $1 AND kind = $2
          ORDER BY id",
    )
    .bind(mission_id)
    .bind(TOOL_CALL)
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|(target, detail)| ToolEvidence {
            tool: target.unwrap_or_default(),
            path: detail
                .get("path")
                .and_then(Value::as_str)
                .map(str::to_string),
            input: detail.get("input").cloned().unwrap_or(Value::Null),
            response: detail.get("response").cloned().unwrap_or(Value::Null),
        })
        .collect())
}

pub async fn record_all(pool: &PgPool, events: Vec<MissionEvent>) {
    for e in events {
        record(pool, e).await;
    }
}

/// The path a tool's **arguments** name, if any.
///
/// Reads the arguments as JSON — never the tool's prose summary. The summary is
/// a sentence written for a human; a path pulled out of it by regex would be
/// right often enough to be trusted and wrong often enough to matter.
///
/// The key names are the ones Claude Code and the ZeroClaw tools actually use.
/// An unrecognised shape returns `None`, which renders as a tool call with no
/// file — accurate, rather than a guess at which argument was a path.
pub fn tool_path(args: &Value) -> Option<String> {
    const KEYS: [&str; 6] = [
        "file_path",
        "filePath",
        "path",
        "notebook_path",
        "file",
        "target_file",
    ];
    let obj = args.as_object()?;
    for k in KEYS {
        if let Some(s) = obj.get(k).and_then(Value::as_str) {
            let s = s.trim();
            if !s.is_empty() {
                return Some(s.to_string());
            }
        }
    }
    None
}

/// Strip the guest/host workspace prefix so a path is repo-relative.
///
/// Tool arguments are absolute inside the sandbox (`/mission/repo/src/a.rs`).
/// Left alone, every mission's file tree would nest under a `mission` → `repo`
/// pair of directory orbs that exist in no repository and mean nothing to the
/// person reading the map.
pub fn repo_relative(path: &str, roots: &[&str]) -> String {
    let p = path.trim();
    for root in roots {
        let root = root.trim_end_matches('/');
        if let Some(rest) = p.strip_prefix(root) {
            let rest = rest.trim_start_matches('/');
            if !rest.is_empty() {
                return rest.to_string();
            }
        }
    }
    p.trim_start_matches("./").to_string()
}

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

    /// Paths come from arguments, and only from argument keys we know.
    ///
    /// The alternative — scanning the values for anything that looks like a
    /// path — is what makes a viz confidently wrong: a `pattern` of `*.rs` or a
    /// `command` of `ls src/` would both become "the agent edited a file".
    #[test]
    fn a_path_comes_from_a_known_argument_or_not_at_all() {
        assert_eq!(
            tool_path(&json!({"file_path": "/mission/repo/src/a.rs"})).as_deref(),
            Some("/mission/repo/src/a.rs")
        );
        assert_eq!(tool_path(&json!({"path": "docs/x.md"})).as_deref(), Some("docs/x.md"));
        // A shell command mentions paths and touches none we can name.
        assert_eq!(tool_path(&json!({"command": "ls src/"})), None);
        // A glob is a query, not a file.
        assert_eq!(tool_path(&json!({"pattern": "**/*.rs"})), None);
        // Blank is absence, not a file called "".
        assert_eq!(tool_path(&json!({"file_path": "   "})), None);
        assert_eq!(tool_path(&json!("not an object")), None);
    }

    /// The sandbox prefix must not become two directory orbs in every mission.
    #[test]
    fn paths_are_made_repo_relative() {
        let roots = ["/mission/repo", "/workspace"];
        assert_eq!(repo_relative("/mission/repo/src/a.rs", &roots), "src/a.rs");
        assert_eq!(repo_relative("/workspace/README.md", &roots), "README.md");
        assert_eq!(repo_relative("./src/a.rs", &roots), "src/a.rs");
        // Outside every root, it is left alone rather than mangled.
        assert_eq!(repo_relative("/etc/hosts", &roots), "/etc/hosts");
        // The root ITSELF is not a file, so it must not collapse to "".
        assert_eq!(repo_relative("/mission/repo", &roots), "/mission/repo");
    }

    /// The cap must be decided inside the INSERT.
    ///
    /// A count-then-insert is the classic version of this and it is wrong here:
    /// the container tap and the microVM drain both write for the same phase,
    /// and each would see a count below the cap and insert. Nothing errors —
    /// the table simply grows past the bound that exists to hold it.
    #[test]
    fn the_cap_is_enforced_in_one_statement() {
        let src = include_str!("mission_events.rs");
        let body = src
            .split("pub async fn record(")
            .nth(1)
            .and_then(|s| s.split("pub async fn").next())
            .expect("record body");
        assert!(
            body.contains("INSERT INTO mission_events") && body.contains("SELECT count(*)"),
            "the cap must be a subquery in the INSERT, not a separate read"
        );
    }
}