tools.rs 90.0%

370 of 411 lines run, 70 of 148 functions · run #19 on main · run missed
//! The tools agents call (spec §15.2). Each is a plain function, so the
//! list is fixed and deterministic, which keeps clients' prompt caches warm.
//! A failure is a result with `isError`, never a protocol error, so the
//! 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 futures::future::BoxFuture;
use serde_json::{json, Value as Json};
use spectra_core::authn::Caller;
use spectra_core::conversation::{self as conv, Author, Post, Posted, Signer};
use spectra_core::daemon::Daemon;
use spectra_core::event::NewEvent;
use spectra_core::ids::now_ms;
use spectra_core::principals::Principal;
use std::sync::Arc;
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;

/// What a tool sees: the daemon and who called it.
pub struct Ctx {
    pub daemon: Arc<Daemon>,
    pub caller: Caller,
    pub session: Option<String>,
}

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
    }
}

/// A tool's result: text for the model, the same as JSON, and whether it
/// failed.
#[derive(Clone, Debug, Default)]
pub struct Output {
    pub text: String,
    pub structured: Option<Json>,
    pub is_error: bool,
    /// An empty `wait`: not recorded (spec §9.2).
    pub quiet: bool,
}

impl Output {
    pub fn ok(text: impl Into<String>, structured: Json) -> Output {
        Output { text: text.into(), structured: Some(structured), ..Default::default() }
    }
    pub fn err(text: impl Into<String>) -> Output {
        Output { text: text.into(), is_error: true, ..Default::default() }
    }
}

fn fail(e: anyhow::Error) -> Output {
    Output::err(format!("{e:#}"))
}

pub type Run = for<'a> fn(&'a Ctx, Json) -> BoxFuture<'a, Output>;

/// One tool as `tools/list` describes it.
pub struct Tool {
    pub name: &'static str,
    pub description: &'static str,
    pub schema: fn() -> Json,
    pub read_only: bool,
    pub destructive: bool,
    /// Answered as an SSE stream with keep-alives (`wait`).
    pub long: bool,
    pub run: Run,
}

impl Tool {
    pub fn describe(&self) -> Json {
        json!({
            "name": self.name,
            "description": self.description,
            "inputSchema": (self.schema)(),
            "annotations": { "readOnlyHint": self.read_only, "destructiveHint": self.destructive },
        })
    }
}

/// Every tool, in the order `tools/list` gives them.
pub fn all() -> &'static [Tool] {
    TOOLS
}

pub fn find(name: &str) -> Option<&'static Tool> {
    TOOLS.iter().find(|t| t.name == name)
}

/// 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),
        }
    })
}

fn agents(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
    Box::pin(async move {
        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
        {
            Ok(r) => r,
            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()
        };
        let list: Vec<Json> = rows
            .iter()
            .map(|(name, owner, kind, project, presence, status)| {
                json!({ "name": name, "owner": owner, "kind": kind, "project": project, "presence": presence.clone().unwrap_or_else(|| "offline".into()), "status": status })
            })
            .collect();
        let text = if rows.is_empty() {
            "No agents.".to_string()
        } else {
            rows.iter()
                .map(|(name, owner, kind, project, presence, status)| {
                    let st = status.as_deref().map(|s| format!(" — \"{s}\"")).unwrap_or_default();
                    format!("- {name} · {owner}'s agent ({kind}, #{project}): {}{st}", presence.as_deref().unwrap_or("offline"))
                })
                .collect::<Vec<_>>()
                .join("\n")
        };
        Output::ok(text, json!({ "agents": list }))
    })
}

/// 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),
        }
    })
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/spectra-mcp/src/tools.rs 90.0%

Keyboard
