The 0.1 tools: brief, rules, agents, status, say, read, inbox, wait with the idle guard, ack, search, redact; every call on the record (#8) 037c45d8093c

-- The rules version each agent last read (spec §12.3), and the tool calls

-- each agent made (spec §9.2), for the agent pages and the idle guard.

CREATE TABLE rules_read (

    agent    TEXT PRIMARY KEY,

    version  TEXT NOT NULL,

    at_ms    INTEGER NOT NULL

);

CREATE TABLE tool_calls (

    seq       INTEGER PRIMARY KEY,

    agent     TEXT NOT NULL,

    tool      TEXT NOT NULL,

    args      TEXT NOT NULL,     -- BLAKE3 of the arguments, kept under attachments/

    result    TEXT NOT NULL,     -- BLAKE3 of the result

    ms        INTEGER NOT NULL,

    is_error  INTEGER NOT NULL,

    read      INTEGER NOT NULL,

    session   TEXT,

    at_ms     INTEGER NOT NULL

);

CREATE INDEX tool_calls_agent ON tool_calls(agent, seq);

//! Bytes kept beside the ledger, named by their BLAKE3 (spec §7.2): tool

//! call arguments and results in full, so the audit is never truncated;

//! retention drops old bodies and keeps the hashes (spec §22.1).

use anyhow::Result;

use std::path::Path;

/// Store `bytes` under `<data>/attachments/` and return their hash (hex).

pub fn put(data: &Path, bytes: &[u8]) -> Result<String> {

    let hash = blake3::hash(bytes).to_hex().to_string();

    let dir = data.join("attachments").join(&hash[..2]);

    let path = dir.join(&hash);

    if !path.exists() {

        std::fs::create_dir_all(&dir)?;

        oxy_core::durable::write_atomic(&path, bytes)?;

    }

    Ok(hash)

}

/// The bytes stored under `hash`, if they are still kept.

pub fn get(data: &Path, hash: &str) -> Option<Vec<u8>> {

    if hash.len() < 3 || !hash.bytes().all(|b| b.is_ascii_hexdigit()) {

        return None;

    }

    std::fs::read(data.join("attachments").join(&hash[..2]).join(hash)).ok()

}

//! Beacon (spec §12): the house rules an agent receives, compiled from the

//! layers that exist, each under a heading naming it, with no other change;

//! the version is the BLAKE3 of that text. 0.1 has the site layer; kind,

//! project and agent layers, and committing them to the ledger, are 0.3.

use crate::event::Event;

use anyhow::{bail, Result};

use std::path::Path;

/// A rule file over this is refused (spec §12.2).

pub const MAX_RULES: usize = 64 * 1024;

/// Compiled rules and their version.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Rules {

    pub text: String,

    /// The first 12 hex characters of BLAKE3(text).

    pub version: String,

}

/// The rules for an agent, from `<data>/rules/`.

pub fn compile(data: &Path) -> Result<Rules> {

    let mut text = String::new();

    let site = data.join("rules").join("site.md");

    if let Ok(body) = std::fs::read_to_string(&site) {

        if body.len() > MAX_RULES {

            bail!("{} is over 64 KiB: the rules must stay readable in one call", site.display());

        }

        text.push_str("# House rules: site (rules/site.md)\n\n");

        text.push_str(body.trim_end());

        text.push('\n');

    }

    let version = blake3::hash(text.as_bytes()).to_hex()[..12].to_string();

    Ok(Rules { text, version })

}

/// The version `agent` last read, if any.

pub async fn last_read(pool: &sqlx::SqlitePool, agent: &str) -> sqlx::Result<Option<String>> {

    sqlx::query_scalar("SELECT version FROM rules_read WHERE agent = ?").bind(agent).fetch_optional(pool).await

}

/// The line every result begins with while an agent works on stale rules.

pub fn changed_line(read: &str, now: &str) -> String {

    format!("House rules changed ({read} → {now}): call rules before continuing.")

}

/// Apply `rules.read` and `tool.called` to their tables.

pub async fn project(conn: &mut sqlx::SqliteConnection, e: &Event) -> sqlx::Result<()> {

    let p = e.json();

    let agent = e.actor.strip_prefix("agent:");

    match e.kind.as_str() {

        "rules.read" => {

            if let (Some(a), Some(v)) = (agent, p.get("version").and_then(|v| v.as_str())) {

                sqlx::query("INSERT INTO rules_read (agent, version, at_ms) VALUES (?, ?, ?) ON CONFLICT(agent) DO UPDATE SET version = excluded.version, at_ms = excluded.at_ms")

                    .bind(a)

                    .bind(v)

                    .bind(e.at_ms as i64)

                    .execute(&mut *conn)

                    .await?;

            }

        }

        "tool.called" => {

            let s = |k: &str| p.get(k).and_then(|v| v.as_str()).map(String::from);

            if let (Some(a), Some(tool)) = (agent, s("tool")) {

                sqlx::query("INSERT OR IGNORE INTO tool_calls (seq, agent, tool, args, result, ms, is_error, read, session, at_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")

                    .bind(e.seq as i64)

                    .bind(a)

                    .bind(tool)

                    .bind(s("args").unwrap_or_default())

                    .bind(s("result").unwrap_or_default())

                    .bind(p.get("ms").and_then(|v| v.as_i64()).unwrap_or(0))

                    .bind(p.get("is_error").and_then(|v| v.as_bool()).unwrap_or(false))

                    .bind(p.get("read").and_then(|v| v.as_bool()).unwrap_or(false))

                    .bind(&e.session)

                    .bind(e.at_ms as i64)

                    .execute(&mut *conn)

                    .await?;

            }

        }

        _ => {}

    }

    Ok(())

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn the_version_follows_the_text_exactly() {

        let dir = tempfile::tempdir().unwrap();

        let empty = compile(dir.path()).unwrap();

        assert_eq!(empty.text, "");

        std::fs::create_dir_all(dir.path().join("rules")).unwrap();

        std::fs::write(dir.path().join("rules/site.md"), "Commit with oxy.\n").unwrap();

        let a = compile(dir.path()).unwrap();

        assert!(a.text.contains("Commit with oxy.") && a.text.starts_with("# House rules: site"));

        std::fs::write(dir.path().join("rules/site.md"), "Commit with oxy!\n").unwrap();

        let b = compile(dir.path()).unwrap();

        assert_ne!(a.version, b.version);

        assert_eq!(a.version.len(), 12);

        std::fs::write(dir.path().join("rules/site.md"), "x".repeat(MAX_RULES + 1)).unwrap();

        assert!(compile(dir.path()).is_err());

    }

}

//! Helpers for end-to-end tests: a hermetic environment, the owner's

//! delegation as `oxy identity delegate add` makes it, agents added through

//! the CLI, and the daemon started and stopped as a real process.

#![allow(dead_code)]

use oxy_core::durable::{write_atomic, write_atomic_private};

use oxy_core::identity::{Delegate, SecretKey};

use spectra_core::keys::KeyPair;

use spectra_core::principals::DELEGATION_SCOPE;

use std::path::{Path, PathBuf};

use std::process::{Child, Command, Output};

pub fn hermetic(dir: &Path) -> Command {

    let mut c = Command::new(env!("CARGO_BIN_EXE_spectra"));

    c.env("SPECTRA_DATA", dir.join("data"))

        .env("OXY_HOME", dir.join("owner"))

        .env("HOME", dir)

        .env("XDG_DATA_HOME", dir.join("xdg"))

        .env("NO_COLOR", "1");

    c

}

pub fn run(dir: &Path, args: &[&str]) -> Output {

    hermetic(dir).args(args).output().unwrap()

}

pub fn text(o: &Output) -> String {

    format!("{}{}", String::from_utf8_lossy(&o.stdout), String::from_utf8_lossy(&o.stderr))

}

/// `oxy identity delegate add` in the owner's home.

pub fn delegate(owner_home: &Path, agent: &KeyPair) {

    let cur = KeyPair::load_home(owner_home).unwrap();

    let keys = owner_home.join("keys");

    let (_, secret) = SecretKey::decode(&std::fs::read(keys.join(format!("{}.secret", cur.key.to_hex()))).unwrap()).unwrap();

    let mut d = cur.identity.delegates.clone();

    d.push(Delegate { key: agent.key, scope: DELEGATION_SCOPE, expires: None });

    let next = cur.identity.successor(d, None, spectra_core::ids::now_ms());

    let k = next.hash();

    write_atomic_private(&keys.join(format!("{}.secret", k.to_hex())), &secret.encode(&k)).unwrap();

    write_atomic(&keys.join(format!("{}.identity", k.to_hex())), &next.encode()).unwrap();

    oxy_core::repo::Config::set_user(&owner_home.join("config.toml"), "author.key", &k.to_hex()).unwrap();

}

/// Make an agent's home, delegate to it, and add it.

pub fn add_agent(d: &Path, name: &str, extra: &[&str]) -> KeyPair {

    let home = d.join("agents").join(name);

    let pair = KeyPair::create_home(&home, name, "").unwrap();

    delegate(&d.join("owner"), &pair);

    let mut args =

        vec!["agent", "add", name, "--kind", "claude-code", "--project", "oxygen", "--home", home.to_str().unwrap(), "--wait", "0"];

    args.extend_from_slice(extra);

    let o = run(d, &args);

    assert!(o.status.success(), "{}", text(&o));

    pair

}

pub struct Serving {

    pub child: Child,

    pub socket: PathBuf,

}

impl Serving {

    pub fn start(d: &Path) -> Serving {

        let mut child = hermetic(d).args(["serve", "--bind", "127.0.0.1:0"]).stdout(std::process::Stdio::piped()).spawn().unwrap();

        let socket = d.join("data/spectra.sock");

        for _ in 0..200 {

            if std::os::unix::net::UnixStream::connect(&socket).is_ok() {

                return Serving { child, socket };

            }

            std::thread::sleep(std::time::Duration::from_millis(25));

        }

        child.kill().ok();

        child.wait().ok();

        panic!("spectra serve did not start");

    }

    pub fn stop(mut self) {

        unsafe { libc::kill(self.child.id() as i32, libc::SIGTERM) };

        let status = self.child.wait().unwrap();

        assert!(status.success(), "a clean stop on SIGTERM");

        assert!(!self.socket.exists(), "the socket is removed");

    }

}

pub fn setup() -> tempfile::TempDir {

    let dir = tempfile::tempdir().unwrap();

    let d = dir.path();

    KeyPair::create_home(&d.join("owner"), "Andy McCall", "").unwrap();

    assert!(run(d, &["init"]).status.success());

    std::fs::create_dir_all(d.join("oxygen")).unwrap();

    assert!(run(d, &["project", "add", d.join("oxygen").to_str().unwrap()]).status.success());

    dir

}

//! The 0.1 tools end to end (spec §15.2, §14.1, §12.3, §9.2): two agents

//! talk through the real daemon, `wait` wakes at once and refuses an idle

//! loop, the rules line appears until the rules are read, every call is

//! recorded with its arguments in full, and the ledger verifies.

mod common;

use common::{add_agent, run, setup, Serving};

use serde_json::{json, Value};

use spectra_core::keys::KeyPair;

use spectra_mcp::client::{rpc_result, Client};

use std::time::{Duration, Instant};

async fn call(c: &Client, name: &str, args: Value) -> Value {

    let req = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": name, "arguments": args,

        "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28" } } });

    let r = c.mcp("2026-07-28", &req).await.unwrap();

    assert_eq!(r.status, 200, "{}", String::from_utf8_lossy(&r.body));

    rpc_result(&r).unwrap()["result"].clone()

}

fn text(v: &Value) -> String {

    v["content"][0]["text"].as_str().unwrap_or("").to_string()

}

#[tokio::test]

async fn two_agents_talk_and_every_call_is_on_the_record() {

    let dir = setup();

    let d = dir.path();

    std::fs::create_dir_all(d.join("data/rules")).unwrap();

    std::fs::write(d.join("data/rules/site.md"), "Commit with oxy, never git.\n").unwrap();

    let claude: KeyPair = add_agent(d, "claude-oxygen", &[]);

    let kimi: KeyPair = add_agent(d, "kimi-oxygen", &[]);

    let srv = Serving::start(d);

    let (c, k) = (Client::new(&srv.socket, Some(claude)), Client::new(&srv.socket, Some(kimi)));

    let b = call(&c, "brief", json!({})).await;

    let t = text(&b);

    assert!(t.contains("You are claude-oxygen · Andy McCall's agent") && t.contains("Commit with oxy, never git."), "{t}");

    assert!(t.contains("kimi-oxygen · Andy McCall's agent"), "{t}");

    let version = b["structuredContent"]["rules"]["version"].as_str().unwrap().to_string();

    call(&k, "brief", json!({})).await;

    // kimi waits; claude mentions it; kimi wakes at once.

    let waiter = tokio::spawn({

        let k = Client::new(&srv.socket, Some(KeyPair::load_home(&d.join("agents/kimi-oxygen")).unwrap()));

        async move {

            let started = Instant::now();

            let r = call(&k, "wait", json!({ "timeout_s": 20 })).await;

            (r, started.elapsed())

        }

    });

    tokio::time::sleep(Duration::from_millis(300)).await;

    let s = call(&c, "say", json!({ "body": "@kimi-oxygen the wire protocol grew a field", "channel": "general" })).await;

    let id = s["structuredContent"]["id"].as_str().unwrap().to_string();

    let (woke, waited) = waiter.await.unwrap();

    assert!(

        text(&woke).contains("the wire protocol grew a field") && text(&woke).contains("[mention from claude-oxygen (Andy McCall's agent"),

        "{}",

        text(&woke)

    );

    assert!(waited < Duration::from_secs(5), "woke after {waited:?}, not at its 20 s timeout");

    let r = call(&k, "read", json!({ "thread": id })).await;

    assert!(text(&r).contains("[from claude-oxygen (Andy McCall's agent, claude-code)") && text(&r).contains("[end]"), "{}", text(&r));

    call(&k, "say", json!({ "body": "thanks, bumping the pin", "reply_to": id })).await;

    call(&k, "ack", json!({ "ids": [id], "reaction": "ack" })).await;

    assert!(text(&call(&k, "inbox", json!({})).await).contains("Nothing unread"));

    let hits = call(&c, "search", json!({ "query": "pin", "since": "1d" })).await;

    assert_eq!(hits["structuredContent"]["hits"].as_array().unwrap().len(), 1, "{hits}");

    call(&c, "status", json!({ "state": "working", "text": "reviewing the field" })).await;

    let a = call(&k, "agents", json!({})).await;

    assert!(text(&a).contains("busy — \"reviewing the field\""), "{}", text(&a));

    // The rules change: the next result says so until `rules` is called.

    std::fs::write(d.join("data/rules/site.md"), "Commit with oxy, never git. Add a changelog line.\n").unwrap();

    let r = call(&c, "inbox", json!({})).await;

    assert!(text(&r).starts_with(&format!("House rules changed ({version} → ")), "{}", text(&r));

    let rr = call(&c, "rules", json!({})).await;

    assert!(text(&rr).contains("Add a changelog line."));

    assert!(!text(&call(&c, "inbox", json!({})).await).starts_with("House rules changed"));

    // Redaction and a refusal come back as results, never protocol errors.

    let mine = call(&c, "say", json!({ "body": "oops, a secret" })).await["structuredContent"]["id"].as_str().unwrap().to_string();

    assert_eq!(call(&k, "redact", json!({ "id": mine, "reason": "not yours" })).await["isError"], true);

    assert_eq!(call(&c, "redact", json!({ "id": mine, "reason": "pasted by mistake" })).await["isError"], false);

    assert_eq!(call(&c, "say", json!({ "body": "x", "channel": "system" })).await["isError"], true);

    srv.stop();

    let v: Value = serde_json::from_slice(&run(d, &["admin", "verify", "--json"]).stdout).unwrap();

    assert!(v["findings"].as_array().unwrap().is_empty(), "{v}");

    let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", d.join("data/spectra.db").display())).await.unwrap();

    let calls: Vec<(String, String)> =

        sqlx::query_as("SELECT tool, args FROM tool_calls WHERE agent = 'claude-oxygen' ORDER BY seq").fetch_all(&pool).await.unwrap();

    assert!(calls.iter().any(|(t, _)| t == "say"));

    let (_, args) = calls.iter().find(|(t, _)| t == "say").unwrap();

    let stored = spectra_core::attachments::get(&d.join("data"), args).expect("arguments kept in full");

    assert!(String::from_utf8_lossy(&stored).contains("the wire protocol grew a field"));

}

#[tokio::test]

async fn an_idle_wait_loop_is_refused() {

    let dir = setup();

    let d = dir.path();

    let a = add_agent(d, "idler", &[]);

    let srv = Serving::start(d);

    let c = Client::new(&srv.socket, Some(a));

    for _ in 0..3 {

        let r = call(&c, "wait", json!({ "timeout_s": 1 })).await;

        assert_eq!((r["isError"].as_bool(), text(&r).as_str()), (Some(false), "Nothing new."));

    }

    let started = Instant::now();

    let r = call(&c, "wait", json!({ "timeout_s": 50 })).await;

    assert_eq!(r["isError"], true);

    assert!(text(&r).contains("End your turn"), "{}", text(&r));

    assert!(started.elapsed() < Duration::from_secs(2), "refused at once");

    srv.stop();

    let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", d.join("data/spectra.db").display())).await.unwrap();

    let waits: i64 =

        sqlx::query_scalar("SELECT COUNT(*) FROM tool_calls WHERE tool = 'wait' AND is_error = 0").fetch_one(&pool).await.unwrap();

    assert_eq!(waits, 0, "empty waits are not recorded");

}

#[tokio::test]

async fn the_tool_list_fits_its_token_budget() {

    let dir = setup();

    let d = dir.path();

    let a = add_agent(d, "counter", &[]);

    let srv = Serving::start(d);

    let c = Client::new(&srv.socket, Some(a));

    let r = c.mcp("2026-07-28", &json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} })).await.unwrap();

    let tools = serde_json::to_string(&r.json().unwrap()["result"]["tools"]).unwrap();

    // About four bytes a token: 1,500 tokens (spec §15.2).

    assert!(tools.len() <= 6_000, "tools/list is {} bytes, about {} tokens", tools.len(), tools.len() / 4);

    srv.stop();

}

- **Added: the 0.1 tools (#8).** `brief`, `rules`, `agents`, `status`, `say`, `read`, `inbox`, `wait`, `ack`, `search` and `redact`, in a `tools/list` under its 1,500-token budget. Every call is recorded as `tool.called` with its arguments and result kept in full, content addressed. While an agent works on stale house rules, every result begins by saying so until it calls `rules`. Messages come back delimited with their author. `wait` wakes as soon as something arrives, and after three empty waits in a row it refuses for ten minutes: an idle agent ends its turn instead of burning tokens. `spectra serve` now stops within five seconds of SIGTERM whatever streams are open.

spectra-mcp.workspace = true

    /// Per agent: empty `wait`s in a row, and until when `wait` is refused.

    pub idle_waits: Mutex<std::collections::HashMap<String, (u32, u64)>>,

            idle_waits: Mutex::new(Default::default()),

/// A time as people and models read it: `2026-11-02 21:43Z`, UTC.

pub fn format_time(ms: u64) -> String {

    let secs = (ms / 1000) as i64;

    let days = secs.div_euclid(86_400);

    let rem = secs.rem_euclid(86_400);

    // Civil from days (Howard Hinnant), as oxy's `format::when`.

    let z = days + 719_468;

    let era = z.div_euclid(146_097);

    let doe = z - era * 146_097;

    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;

    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);

    let mp = (5 * doy + 2) / 153;

    let d = doy - (153 * mp + 2) / 5 + 1;

    let m = if mp < 10 { mp + 3 } else { mp - 9 };

    let y = yoe + era * 400 + if m <= 2 { 1 } else { 0 };

    format!("{y:04}-{m:02}-{d:02} {:02}:{:02}Z", rem / 3600, (rem % 3600) / 60)

}

        assert_eq!(format_time(1_793_655_780_000), "2026-11-02 21:43Z");

pub mod attachments;

pub mod rules;

    crate::rules::project(conn, e).await?;

        let mut req = hyper::Request::builder()

            .method(method)

            .uri(path)

            .header("host", "spectra")

            .header("content-type", "application/json")

            .header("connection", "close");

        // One request a connection: the server closes it after answering, so

        // nothing lingers to hold up its shutdown.

                let out = crate::tools::invoke(&ctx, tool, args).await;

/// `_meta` on every result (spec §15.2): the rules version, the session and

/// the unread count.

    if let Ok(r) = spectra_core::rules::compile(&ctx.daemon.data) {

        m.insert("spectra/rules_version".into(), json!(r.version));

    }

    let _ = &meta;

            let out = tools::invoke(ctx, tool, args).await;

            let meta = crate::http::meta(ctx).await;

//! model sees why. Every call goes through [`invoke`]: it is recorded as

//! `tool.called` with its arguments and result in full (content addressed),

//! and its text begins with the rules-change line while the caller works on

//! stale rules (spec §12.3).

//!

//! Text from other principals is always delimited with its author (spec

//! §20.3), so a model can tell whose words it is reading.

use spectra_core::conversation::{self as conv, Author, Post, Posted, Signer};

use spectra_core::event::NewEvent;

use spectra_core::ids::now_ms;

use std::time::Duration;

/// Empty `wait`s in a row before `wait` is refused, and for how long.

pub const IDLE_WAITS: u32 = 3;

pub const IDLE_COOLDOWN_MS: u64 = 10 * 60_000;

impl Ctx {

    fn author(&self) -> Author<'static> {

        Author { principal: self.caller.principal.clone(), signer: Signer::Request(self.caller.auth.clone()) }

    }

    fn actor(&self) -> String {

        self.caller.principal.actor()

    }

    fn event(&self, kind: &str, payload: Json) -> NewEvent {

        let mut e = NewEvent::system(kind, payload).by(self.actor(), self.caller.auth.clone());

        e.session = self.session.clone();

        e

    }

}

    /// An empty `wait`: not recorded (spec §9.2).

    pub quiet: bool,

        Output { text: text.into(), structured: Some(structured), ..Default::default() }

        Output { text: text.into(), is_error: true, ..Default::default() }

}

fn fail(e: anyhow::Error) -> Output {

    Output::err(format!("{e:#}"))

}

/// Run a tool for the caller: the idle guard, the rules line, and the record.

pub async fn invoke(ctx: &Ctx, tool: &Tool, args: Json) -> Output {

    let started = std::time::Instant::now();

    let mut out = (tool.run)(ctx, args.clone()).await;

    let Principal::Agent { name, .. } = &ctx.caller.principal else { return out };

    if tool.name != "rules" && tool.name != "brief" {

        // With no house rules there is nothing to read, so nothing to say.

        if let Some(rules) = spectra_core::rules::compile(&ctx.daemon.data).ok().filter(|r| !r.text.is_empty()) {

            let read = spectra_core::rules::last_read(ctx.daemon.ledger.pool(), name).await.ok().flatten();

            if read.as_deref() != Some(rules.version.as_str()) {

                let from = read.unwrap_or_else(|| "none".into());

                out.text = format!("{}\n\n{}", spectra_core::rules::changed_line(&from, &rules.version), out.text);

            }

        }

    }

    if out.quiet {

        return out;

    }

    let data = &ctx.daemon.data;

    let args_hash = spectra_core::attachments::put(data, args.to_string().as_bytes()).unwrap_or_default();

    let result_hash = spectra_core::attachments::put(data, out.text.as_bytes()).unwrap_or_default();

    let e = ctx.event(

        "tool.called",

        json!({ "tool": tool.name, "args": args_hash, "result": result_hash, "ms": started.elapsed().as_millis() as u64, "is_error": out.is_error, "read": tool.read_only }),

    );

    let _ = ctx.daemon.ledger.append_one(e).await;

    out

}

fn obj(props: Json, required: &[&str]) -> Json {

    json!({ "type": "object", "properties": props, "required": required, "additionalProperties": false })

}

static TOOLS: &[Tool] = &[

    Tool {

        name: "brief",

        description: "Call first: who you are, your project, the house rules, your unread inbox, and who else is here.",

        schema: || obj(json!({ "new_session": { "type": "boolean" } }), &[]),

        read_only: true,

        destructive: false,

        long: false,

        run: brief,

    },

    Tool {

        name: "rules",

        description: "The house rules and their version, in full.",

        schema: || obj(json!({}), &[]),

        read_only: true,

        destructive: false,

        long: false,

        run: rules,

    },

    Tool {

        name: "agents",

        description: "Every agent with its owner, kind, project, presence and status.",

        schema: || obj(json!({ "project": { "type": "string" } }), &[]),

        read_only: true,

        destructive: false,

        long: false,

        run: agents,

    },

    Tool {

        name: "status",

        description: "Say what you are doing, so others can see.",

        schema: || obj(json!({ "state": { "enum": ["working", "idle", "blocked"] }, "text": { "type": "string" }, "task": { "type": "string" } }), &[]),

        read_only: false,

        destructive: false,

        long: false,

        run: status,

    },

    Tool {

        name: "say",

        description: "Post a message (default: your project's channel). @name mentions; reply_to answers in a thread.",

        schema: || {

            obj(

                json!({ "body": { "type": "string" }, "channel": { "type": "string" }, "reply_to": { "type": "string" },

                    "kind": { "enum": ["chat", "request", "answer", "handoff", "status", "decision"] }, "to": { "type": "string" } }),

                &["body"],

            )

        },

        read_only: false,

        destructive: false,

        long: false,

        run: say,

    },

    Tool {

        name: "read",

        description: "Messages of a channel (default: your project's) or a whole thread; marks nothing read.",

        schema: || obj(json!({ "channel": { "type": "string" }, "thread": { "type": "string" }, "since": { "type": "string" }, "limit": { "type": "integer" } }), &[]),

        read_only: true,

        destructive: false,

        long: false,

        run: read,

    },

    Tool {

        name: "inbox",

        description: "Your unread mentions and direct messages.",

        schema: || obj(json!({}), &[]),

        read_only: true,

        destructive: false,

        long: false,

        run: inbox,

    },

    Tool {

        name: "wait",

        description: "Only while waiting for an answer inside a piece of work: blocks up to timeout_s for something new. If idle, end your turn instead.",

        schema: || obj(json!({ "timeout_s": { "type": "integer", "minimum": 1, "maximum": 50 } }), &["timeout_s"]),

        read_only: true,

        destructive: false,

        long: true,

        run: wait,

    },

    Tool {

        name: "ack",

        description: "Mark inbox items read, with an optional reaction (ack, done, eyes, no).",

        schema: || obj(json!({ "ids": { "type": "array", "items": { "type": "string" } }, "reaction": { "enum": ["ack", "done", "eyes", "no"] } }), &["ids"]),

        read_only: false,

        destructive: false,

        long: false,

        run: ack,

    },

    Tool {

        name: "search",

        description: "Search history: words, \"phrases\", prefix*, AND/OR/NOT; filter by channel, author, since (7d, 3h).",

        schema: || {

            obj(json!({ "query": { "type": "string" }, "channel": { "type": "string" }, "author": { "type": "string" }, "since": { "type": "string" }, "limit": { "type": "integer" } }), &["query"])

        },

        read_only: true,

        destructive: false,

        long: false,

        run: search,

    },

    Tool {

        name: "redact",

        description: "Redact your own message, with the reason.",

        schema: || obj(json!({ "id": { "type": "string" }, "reason": { "type": "string" } }), &["id", "reason"]),

        read_only: false,

        destructive: true,

        long: false,

        run: redact,

    },

];

fn s<'a>(args: &'a Json, k: &str) -> Option<&'a str> {

    args.get(k).and_then(Json::as_str).filter(|v| !v.is_empty())

}

/// How a principal is named to a model: `claude-oxygen (Andy McCall's agent, claude-code)`.

async fn label(d: &Daemon, actor: &str) -> String {

    if actor == "human" {

        let name: Option<String> = sqlx::query_scalar("SELECT name FROM identities WHERE role = 'owner' AND current = 1")

            .fetch_optional(d.ledger.pool())

            .await

            .ok()

            .flatten();

        return format!("{} (the human)", name.unwrap_or_else(|| "the owner".into()));

    }

    if let Some(a) = actor.strip_prefix("agent:") {

        let row: Option<(String, String)> = sqlx::query_as("SELECT owner_name, kind FROM agents WHERE name = ?")

            .bind(a)

            .fetch_optional(d.ledger.pool())

            .await

            .ok()

            .flatten();

        if let Some((owner, kind)) = row {

            return format!("{a} ({owner}'s agent, {kind})");

        }

        return a.to_string();

    }

    "Spectra".into()

}

/// A message as a model reads it: delimited with its author (spec §20.3).

async fn delimited(d: &Daemon, m: &conv::Message, channel: &str) -> String {

    let at = spectra_core::ids::format_time(m.at_ms as u64);

    let thread = m.reply_to.as_deref().map(|r| format!(", reply in thread {r}")).unwrap_or_default();

    let replies = if m.replies > 0 { format!(" ({} replies)", m.replies) } else { String::new() };

    format!("[from {} at {at} in {channel}, id {}{thread}{replies}]\n{}\n[end]", label(d, &m.author).await, m.id, m.body)

fn brief(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    let _ = args;

    Box::pin(async move {

        let d = &ctx.daemon;

        let rules = match spectra_core::rules::compile(&d.data) {

            Ok(r) => r,

            Err(e) => return fail(e),

        };

        let me = ctx.caller.principal.display();

        let project = match &ctx.caller.principal {

            Principal::Agent { project, .. } => Some(project.clone()),

            _ => None,

        };

        let mut relations = (Vec::<String>::new(), Vec::<String>::new());

        if let Some(p) = &project {

            relations.0 = sqlx::query_scalar("SELECT depends_on FROM project_relations WHERE project = ?")

                .bind(p)

                .fetch_all(d.ledger.pool())

                .await

                .unwrap_or_default();

            relations.1 = sqlx::query_scalar("SELECT project FROM project_relations WHERE depends_on = ?")

                .bind(p)

                .fetch_all(d.ledger.pool())

                .await

                .unwrap_or_default();

        }

        let items = conv::inbox(&d.ledger, &ctx.actor(), Some(&ctx.author())).await.unwrap_or_default();

        let others = agents(ctx, json!({})).await;

        let _ = d.ledger.append_one(ctx.event("rules.read", json!({ "version": rules.version }))).await;

        let mut text = format!("You are {me}.");

        if let Some(p) = &project {

            text.push_str(&format!(" Your project is {p}; your channel is #{p}."));

            if !relations.0.is_empty() {

                text.push_str(&format!(" It pins {}.", relations.0.join(", ")));

            }

            if !relations.1.is_empty() {

                text.push_str(&format!(" {} pin it.", relations.1.join(", ")));

            }

        }

        if let Some(sess) = &ctx.session {

            text.push_str(&format!(" Session {sess}."));

        }

        text.push_str(&format!("\n\nHouse rules (version {}):\n", rules.version));

        text.push_str(if rules.text.is_empty() { "(none yet)\n" } else { &rules.text });

        text.push_str(&format!("\nInbox: {} unread.\n", items.len()));

        for i in &items {

            text.push_str(&format!("- {} from {} in {} ({}): {}\n", i.kind, label(d, &i.from).await, i.channel, i.id, i.summary));

        }

        text.push_str(&format!("\nAgents:\n{}\n", others.text));

        text.push_str("\nWhen you have nothing to do, end your turn: Spectra starts you again when there is work.");

        Output::ok(

            text,

            json!({ "you": me, "project": project, "session": ctx.session, "rules": { "version": rules.version, "text": rules.text },

                "inbox": items, "agents": others.structured.and_then(|s| s.get("agents").cloned()), "pins": relations.0, "pinned_by": relations.1 }),

        )

    })

}

fn rules(ctx: &Ctx, _args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        match spectra_core::rules::compile(&ctx.daemon.data) {

            Ok(r) => {

                let _ = ctx.daemon.ledger.append_one(ctx.event("rules.read", json!({ "version": r.version }))).await;

                let body = if r.text.is_empty() { "(no house rules yet)".to_string() } else { r.text.clone() };

                Output::ok(format!("House rules, version {}:\n\n{body}", r.version), json!({ "version": r.version, "text": r.text }))

            }

            Err(e) => fail(e),

        }

    })

        let project = s(&args, "project");

        let rows: Vec<AgentRow> = match sqlx::query_as(

            "SELECT a.name, a.owner_name, a.kind, a.project, p.state, p.status_text FROM agents a LEFT JOIN presence p ON p.agent = a.name \

             WHERE a.disabled_ms IS NULL AND (? IS NULL OR a.project = ?) ORDER BY a.name",

        )

        .bind(project)

        .bind(project)

        .fetch_all(ctx.daemon.ledger.pool())

        .await

        {

            Err(e) => return fail(e.into()),

        // Presence as the daemon sees it now; the table is its recorded history.

        let rows: Vec<_> = {

            let p = ctx.daemon.presence.lock().await;

            let now = now_ms();

            rows.into_iter()

                .map(|(name, owner, kind, project, _, status)| {

                    let live = p.state(&name, now).to_string();

                    (name, owner, kind, project, Some(live), status)

                })

                .collect()

        };

    })

}

/// An agent as the `agents` tool reads it: name, owner, kind, project,

/// presence and status.

type AgentRow = (String, String, String, String, Option<String>, Option<String>);

fn status(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let Principal::Agent { name, .. } = &ctx.caller.principal else { return Output::err("status is for agents") };

        let state = s(&args, "state").unwrap_or("working");

        let text = s(&args, "text");

        if text.is_some_and(|t| t.len() > 500) {

            return Output::err("a status is at most 500 characters");

        }

        let e = ctx.event("agent.status", json!({ "state": state, "text": text, "task": s(&args, "task") })).subject(name.clone());

        if let Err(e) = ctx.daemon.ledger.append_one(e).await {

            return fail(e);

        }

        {

            let mut p = ctx.daemon.presence.lock().await;

            p.status(name, (state != "idle").then(now_ms));

            p.blocked(name, state == "blocked");

        }

        Output::ok(

            format!("Status set: {state}{}", text.map(|t| format!(" — {t}")).unwrap_or_default()),

            json!({ "state": state, "text": text }),

        )

    })

}

fn say(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let Some(body) = s(&args, "body") else { return Output::err("say needs a body") };

        let mut channel = s(&args, "channel").map(String::from);

        if let Some(to) = s(&args, "to") {

            match conv::ensure_direct(&ctx.daemon.ledger, &ctx.caller.principal, to).await {

                Ok(c) => channel = Some(c.name),

                Err(e) => return fail(e),

            }

        }

        let post = Post {

            channel,

            body: body.into(),

            reply_to: s(&args, "reply_to").map(String::from),

            kind: s(&args, "kind").map(String::from),

            ..Default::default()

        };

        match conv::post(&ctx.daemon.ledger, &ctx.author(), post, now_ms()).await {

            Ok(Posted::Message { id, channel }) => {

                Output::ok(format!("Posted {id} in {channel}."), json!({ "id": id, "channel": channel }))

            }

            Ok(Posted::Duplicate { id }) => {

                Output::ok(format!("Not posted: you said exactly this in the last minute ({id})."), json!({ "duplicate_of": id }))

            }

            Err(e) => fail(e),

        }

    })

}

fn read(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let limit = args.get("limit").and_then(Json::as_u64).unwrap_or(50) as u32;

        match conv::read(&ctx.daemon.ledger, &ctx.caller.principal, s(&args, "channel"), s(&args, "thread"), s(&args, "since"), limit).await

        {

            Ok((ch, msgs)) => {

                let mut text = format!("{} message(s) in {}:\n\n", msgs.len(), ch.display());

                for m in &msgs {

                    text.push_str(&delimited(&ctx.daemon, m, &ch.display()).await);

                    text.push_str("\n\n");

                }

                Output::ok(text.trim_end().to_string(), json!({ "channel": ch.display(), "messages": msgs }))

            }

            Err(e) => fail(e),

        }

    })

}

fn inbox(ctx: &Ctx, _args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        match conv::inbox(&ctx.daemon.ledger, &ctx.actor(), Some(&ctx.author())).await {

            Ok(items) => inbox_output(ctx, items).await,

            Err(e) => fail(e),

        }

    })

}

async fn inbox_output(ctx: &Ctx, items: Vec<conv::Item>) -> Output {

    if items.is_empty() {

        return Output::ok("Nothing unread.", json!({ "items": [] }));

    }

    let mut text = format!("{} unread:\n\n", items.len());

    for i in &items {

        text.push_str(&format!(

            "[{} from {} in {}, id {}]\n{}\n[end]\nAct on it: {}\n\n",

            i.kind,

            label(&ctx.daemon, &i.from).await,

            i.channel,

            i.id,

            i.summary,

            i.action_hint

        ));

    }

    Output::ok(text.trim_end().to_string(), json!({ "items": items }))

}

fn wait(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let actor = ctx.actor();

        let now = now_ms();

        {

            let guard = ctx.daemon.idle_waits.lock().await;

            if let Some((_, until)) = guard.get(&actor) {

                if *until > now {

                    return Output::err("Nothing is waiting for you. End your turn; Spectra starts you again when there is work.");

                }

            }

        }

        let timeout = args.get("timeout_s").and_then(Json::as_u64).unwrap_or(30).clamp(1, 50);

        let name = actor.strip_prefix("agent:").unwrap_or(&actor).to_string();

        // Subscribe before looking, so nothing is missed between the two.

        let mut rx = ctx.daemon.ledger.subscribe();

        let ledger = &ctx.daemon.ledger;

        ctx.daemon.presence.lock().await.wait_started(&name);

        let found = async {

            loop {

                if conv::unread(ledger, &actor).await.unwrap_or(0) > 0 {

                    return true;

                }

                match rx.recv().await {

                    Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,

                    Err(_) => return false,

                }

            }

        };

        let got = tokio::time::timeout(Duration::from_secs(timeout), found).await.unwrap_or(false);

        ctx.daemon.presence.lock().await.wait_ended(&name);

        let mut guard = ctx.daemon.idle_waits.lock().await;

        if got {

            guard.remove(&actor);

            drop(guard);

            return match conv::inbox(ledger, &actor, Some(&ctx.author())).await {

                Ok(items) => inbox_output(ctx, items).await,

                Err(e) => fail(e),

            };

        }

        let entry = guard.entry(actor).or_insert((0, 0));

        entry.0 += 1;

        if entry.0 >= IDLE_WAITS {

            entry.1 = now_ms() + IDLE_COOLDOWN_MS;

            entry.0 = 0;

        }

        Output { text: "Nothing new.".into(), structured: Some(json!({ "items": [] })), is_error: false, quiet: true }

    })

}

fn ack(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let ids: Vec<String> = args

            .get("ids")

            .and_then(Json::as_array)

            .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())

            .unwrap_or_default();

        if ids.is_empty() {

            return Output::err("ack needs ids");

        }

        match conv::ack(&ctx.daemon.ledger, &ctx.author(), &ids, s(&args, "reaction")).await {

            Ok(n) => Output::ok(format!("Marked {n} read."), json!({ "read": n })),

            Err(e) => fail(e),

        }

    })

}

fn search(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let Some(q) = s(&args, "query") else { return Output::err("search needs a query") };

        let since = s(&args, "since").and_then(|v| parse_since(v, now_ms()));

        let limit = args.get("limit").and_then(Json::as_u64).unwrap_or(20) as u32;

        match conv::search(&ctx.daemon.ledger, &ctx.caller.principal, q, s(&args, "channel"), s(&args, "author"), since, limit).await {

            Ok(hits) => {

                let mut text = format!("{} hit(s):\n", hits.len());

                for h in &hits {

                    text.push_str(&format!(

                        "- [{} in {} at {}, id {}] {}\n",

                        label(&ctx.daemon, &h.author).await,

                        h.channel,

                        spectra_core::ids::format_time(h.at_ms as u64),

                        h.id,

                        h.snippet

                    ));

                }

                Output::ok(text.trim_end().to_string(), json!({ "hits": hits }))

            }

            Err(e) => fail(e),

        }

/// `7d`, `3h`, `30m`, or milliseconds since the epoch.

pub fn parse_since(v: &str, now: u64) -> Option<i64> {

    let (n, unit) = v.split_at(v.len().saturating_sub(1));

    let ms = match unit {

        "d" => 86_400_000,

        "h" => 3_600_000,

        "m" => 60_000,

        _ => return v.parse().ok(),

    };

    n.parse::<u64>().ok().map(|n| now.saturating_sub(n * ms) as i64)

}

fn redact(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let (Some(id), Some(reason)) = (s(&args, "id"), s(&args, "reason")) else { return Output::err("redact needs id and reason") };

        match conv::redact(&ctx.daemon.ledger, &ctx.author(), id, reason).await {

            Ok(()) => Output::ok(format!("Redacted {id}."), json!({ "id": id })),

            Err(e) => fail(e),

        }

    })

- [x] #8 The 0.1 tools — done 2026-11-02: eleven tools through one invoke wrapper (record, rules line); wait with the idle guard; end-to-end tests with two agents. Found while testing: serve could hang on shutdown with a stream open (now five seconds at most).

    // After the signal, open connections get five seconds to finish: a

    // `subscriptions/listen` stream never does by itself.

    let (stopping_tx, mut stopping) = tokio::sync::watch::channel(false);

    tokio::spawn(async move {

        signal.await;

        let _ = stopping_tx.send(true);

    });

    let servers = async { tokio::join!(unix_srv, tcp_srv) };

    tokio::select! {

        (u, t) = servers => {

            u?;

            t?;

        }

        _ = async {

            let _ = stopping.wait_for(|s| *s).await;

            tokio::time::sleep(std::time::Duration::from_secs(5)).await;

        } => {

            tracing::info!("connections still open five seconds after the signal; stopping anyway");

        }

    }

mod common;

use common::{add_agent, run, setup, text, Serving};