OXYGENHUB

Coverage / crates/spectra-mcp/src/tools.rs 90.0%

377 of 419 lines run, 71 of 150 functions · run #28 on main · run missed

1
//! The tools agents call (spec §15.2). Each is a plain function, so the
2
//! list is fixed and deterministic, which keeps clients' prompt caches warm.
3
//! A failure is a result with `isError`, never a protocol error, so the
4
//! model sees why. Every call goes through [`invoke`]: it is recorded as
5
//! `tool.called` with its arguments and result in full (content addressed),
6
//! and its text begins with the rules-change line while the caller works on
7
//! stale rules (spec §12.3).
8
//!
9
//! Text from other principals is always delimited with its author (spec
10
//! §20.3), so a model can tell whose words it is reading.
11
12
use futures::future::BoxFuture;
13
use serde_json::{json, Value as Json};
14
use spectra_core::authn::Caller;
15
use spectra_core::conversation::{self as conv, Author, Post, Posted, Signer};
16
use spectra_core::daemon::Daemon;
17
use spectra_core::event::NewEvent;
18
use spectra_core::ids::now_ms;
19
use spectra_core::principals::Principal;
20
use std::sync::Arc;
21
use std::time::Duration;
22
23
/// Empty `wait`s in a row before `wait` is refused, and for how long.
24
pub const IDLE_WAITS: u32 = 3;
25
pub const IDLE_COOLDOWN_MS: u64 = 10 * 60_000;
26
27
/// What a tool sees: the daemon and who called it.
28
pub struct Ctx {
29
    pub daemon: Arc<Daemon>,
30
    pub caller: Caller,
31
    pub session: Option<String>,
32
}
33
34
impl Ctx {
3521×
    fn author(&self) -> Author<'static> {
3621×
        Author { principal: self.caller.principal.clone(), signer: Signer::Request(self.caller.auth.clone()) }
3721×
    }
3862×
    fn actor(&self) -> String {
3962×
        self.caller.principal.actor()
4062×
    }
4147×
    fn event(&self, kind: &str, payload: Json) -> NewEvent {
4247×
        let mut e = NewEvent::system(kind, payload).by(self.actor(), self.caller.auth.clone());
4347×
        e.session = self.session.clone();
4447×
        e
4547×
    }
46
}
47
48
/// A tool's result: text for the model, the same as JSON, and whether it
49
/// failed.
50
#[derive(Clone, Debug, Default)]
51
pub struct Output {
52
    pub text: String,
53
    pub structured: Option<Json>,
54
    pub is_error: bool,
55
    /// An empty `wait`: not recorded (spec §9.2).
56
    pub quiet: bool,
57
}
58
59
impl Output {
6036×
    pub fn ok(text: impl Into<String>, structured: Json) -> Output {
6136×
        Output { text: text.into(), structured: Some(structured), ..Default::default() }
6236×
    }
633×
    pub fn err(text: impl Into<String>) -> Output {
643×
        Output { text: text.into(), is_error: true, ..Default::default() }
653×
    }
66
}
67
682×
fn fail(e: anyhow::Error) -> Output {
692×
    Output::err(format!("{e:#}"))
702×
}
71
72
pub type Run = for<'a> fn(&'a Ctx, Json) -> BoxFuture<'a, Output>;
73
74
/// One tool as `tools/list` describes it.
75
pub struct Tool {
76
    pub name: &'static str,
77
    pub description: &'static str,
78
    pub schema: fn() -> Json,
79
    pub read_only: bool,
80
    pub destructive: bool,
81
    /// Answered as an SSE stream with keep-alives (`wait`).
82
    pub long: bool,
83
    pub run: Run,
84
}
85
86
impl Tool {
8755×
    pub fn describe(&self) -> Json {
8855×
        json!({
8955×
            "name": self.name,
9055×
            "description": self.description,
9155×
            "inputSchema": (self.schema)(),
9255×
            "annotations": { "readOnlyHint": self.read_only, "destructiveHint": self.destructive },
93
        })
9455×
    }
95
}
96
97
/// Every tool, in the order `tools/list` gives them.
985×
pub fn all() -> &'static [Tool] {
995×
    TOOLS
1005×
}
101
10238×
pub fn find(name: &str) -> Option<&'static Tool> {
103196×
    TOOLS.iter().find(|t| t.name == name)
10438×
}
105
106
/// Run a tool for the caller: the idle guard, the rules line, and the record.
10738×
pub async fn invoke(ctx: &Ctx, tool: &Tool, args: Json) -> Output {
10838×
    let started = std::time::Instant::now();
10938×
    let mut out = (tool.run)(ctx, args.clone()).await;
11038×
    let Principal::Agent { name, .. } = &ctx.caller.principal else { return out };
11138×
    if tool.name != "rules" && tool.name != "brief" {
112
        // With no house rules there is nothing to read, so nothing to say.
11327×
        if let Some(rules) = spectra_core::rules::compile(ctx.daemon.ledger.pool(), project_of(&ctx.caller.principal))
11427×
            .await
11527×
            .ok()
11627×
            .filter(|r| !r.text.is_empty())
117
        {
11817×
            let read = spectra_core::rules::last_read(ctx.daemon.ledger.pool(), name).await.ok().flatten();
11917×
            if read.as_deref() != Some(rules.version.as_str()) {
1203×
                let from = read.unwrap_or_else(|| "none".into());
1213×
                out.text = format!("{}\n\n{}", spectra_core::rules::changed_line(&from, &rules.version), out.text);
12214×
            }
12310×
        }
12411×
    }
12538×
    if out.quiet {
1263×
        return out;
12735×
    }
12835×
    let data = &ctx.daemon.data;
12935×
    let args_hash = spectra_core::attachments::put(data, args.to_string().as_bytes()).unwrap_or_default();
13035×
    let result_hash = spectra_core::attachments::put(data, out.text.as_bytes()).unwrap_or_default();
13135×
    let e = ctx.event(
13235×
        "tool.called",
13335×
        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 }),
134
    );
13535×
    let _ = ctx.daemon.ledger.append_one(e).await;
13635×
    out
13738×
}
138
13955×
fn obj(props: Json, required: &[&str]) -> Json {
14055×
    json!({ "type": "object", "properties": props, "required": required, "additionalProperties": false })
14155×
}
142
143
static TOOLS: &[Tool] = &[
144
    Tool {
145
        name: "brief",
146
        description: "Call first: who you are, your project, your brief, your unread inbox, and who else is here.",
1475×
        schema: || obj(json!({ "new_session": { "type": "boolean" } }), &[]),
148
        read_only: true,
149
        destructive: false,
150
        long: false,
151
        run: brief,
152
    },
153
    Tool {
154
        name: "rules",
155
        description: "Your brief (overall and your project's) and its version, in full.",
1565×
        schema: || obj(json!({}), &[]),
157
        read_only: true,
158
        destructive: false,
159
        long: false,
160
        run: rules,
161
    },
162
    Tool {
163
        name: "agents",
164
        description: "Every agent with its owner, kind, project, presence and status.",
1655×
        schema: || obj(json!({ "project": { "type": "string" } }), &[]),
166
        read_only: true,
167
        destructive: false,
168
        long: false,
169
        run: agents,
170
    },
171
    Tool {
172
        name: "status",
173
        description: "Say what you are doing, so others can see.",
1745×
        schema: || obj(json!({ "state": { "enum": ["working", "idle", "blocked"] }, "text": { "type": "string" }, "task": { "type": "string" } }), &[]),
175
        read_only: false,
176
        destructive: false,
177
        long: false,
178
        run: status,
179
    },
180
    Tool {
181
        name: "say",
182
        description: "Post a message (default: your project's channel). @name mentions; reply_to answers in a thread.",
1835×
        schema: || {
1845×
            obj(
1855×
                json!({ "body": { "type": "string" }, "channel": { "type": "string" }, "reply_to": { "type": "string" },
1865×
                    "kind": { "enum": ["chat", "request", "answer", "handoff", "status", "decision"] }, "to": { "type": "string" } }),
1875×
                &["body"],
188
            )
1895×
        },
190
        read_only: false,
191
        destructive: false,
192
        long: false,
193
        run: say,
194
    },
195
    Tool {
196
        name: "read",
197
        description: "Messages of a channel (default: your project's) or a whole thread; marks nothing read.",
1985×
        schema: || obj(json!({ "channel": { "type": "string" }, "thread": { "type": "string" }, "since": { "type": "string" }, "limit": { "type": "integer" } }), &[]),
199
        read_only: true,
200
        destructive: false,
201
        long: false,
202
        run: read,
203
    },
204
    Tool {
205
        name: "inbox",
206
        description: "Your unread mentions and direct messages.",
2075×
        schema: || obj(json!({}), &[]),
208
        read_only: true,
209
        destructive: false,
210
        long: false,
211
        run: inbox,
212
    },
213
    Tool {
214
        name: "wait",
215
        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.",
2165×
        schema: || obj(json!({ "timeout_s": { "type": "integer", "minimum": 1, "maximum": 50 } }), &["timeout_s"]),
217
        read_only: true,
218
        destructive: false,
219
        long: true,
220
        run: wait,
221
    },
222
    Tool {
223
        name: "ack",
224
        description: "Mark inbox items read, with an optional reaction (ack, done, eyes, no).",
2255×
        schema: || obj(json!({ "ids": { "type": "array", "items": { "type": "string" } }, "reaction": { "enum": ["ack", "done", "eyes", "no"] } }), &["ids"]),
226
        read_only: false,
227
        destructive: false,
228
        long: false,
229
        run: ack,
230
    },
231
    Tool {
232
        name: "search",
233
        description: "Search history: words, \"phrases\", prefix*, AND/OR/NOT; filter by channel, author, since (7d, 3h).",
2345×
        schema: || {
2355×
            obj(json!({ "query": { "type": "string" }, "channel": { "type": "string" }, "author": { "type": "string" }, "since": { "type": "string" }, "limit": { "type": "integer" } }), &["query"])
2365×
        },
237
        read_only: true,
238
        destructive: false,
239
        long: false,
240
        run: search,
241
    },
242
    Tool {
243
        name: "redact",
244
        description: "Redact your own message, with the reason.",
2455×
        schema: || obj(json!({ "id": { "type": "string" }, "reason": { "type": "string" } }), &["id", "reason"]),
246
        read_only: false,
247
        destructive: true,
248
        long: false,
249
        run: redact,
250
    },
251
];
252
25357×
fn s<'a>(args: &'a Json, k: &str) -> Option<&'a str> {
25457×
    args.get(k).and_then(Json::as_str).filter(|v| !v.is_empty())
25557×
}
256
257
/// How a principal is named to a model: `claude-oxygen (Andy McCall's agent, claude-code)`.
2584×
async fn label(d: &Daemon, actor: &str) -> String {
2594×
    if actor == "human" {
2600×
        let name: Option<String> = sqlx::query_scalar("SELECT name FROM identities WHERE role = 'owner' AND current = 1")
2610×
            .fetch_optional(d.ledger.pool())
2620×
            .await
2630×
            .ok()
2640×
            .flatten();
2650×
        return format!("{} (the human)", name.unwrap_or_else(|| "the owner".into()));
2664×
    }
2674×
    if let Some(a) = actor.strip_prefix("agent:") {
2684×
        let row: Option<(String, String)> = sqlx::query_as("SELECT owner_name, kind FROM agents WHERE name = ?")
2694×
            .bind(a)
2704×
            .fetch_optional(d.ledger.pool())
2714×
            .await
2724×
            .ok()
2734×
            .flatten();
2744×
        if let Some((owner, kind)) = row {
2754×
            return format!("{a} ({owner}'s agent, {kind})");
2760×
        }
2770×
        return a.to_string();
2780×
    }
2790×
    "Spectra".into()
2804×
}
281
282
/// A message as a model reads it: delimited with its author (spec §20.3).
2831×
async fn delimited(d: &Daemon, m: &conv::Message, channel: &str) -> String {
2841×
    let at = spectra_core::ids::format_time(m.at_ms as u64);
2851×
    let thread = m.reply_to.as_deref().map(|r| format!(", reply in thread {r}")).unwrap_or_default();
2861×
    let replies = if m.replies > 0 { format!(" ({} replies)", m.replies) } else { String::new() };
2871×
    format!("[from {} at {at} in {channel}, id {}{thread}{replies}]\n{}\n[end]", label(d, &m.author).await, m.id, m.body)
2881×
}
289
2904×
fn brief(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
2914×
    let _ = args;
2924×
    Box::pin(async move {
2934×
        let d = &ctx.daemon;
2944×
        let rules = match spectra_core::rules::compile(d.ledger.pool(), project_of(&ctx.caller.principal)).await {
2954×
            Ok(r) => r,
2960×
            Err(e) => return fail(e),
297
        };
2984×
        let me = ctx.caller.principal.display();
2994×
        let project = match &ctx.caller.principal {
3004×
            Principal::Agent { project, .. } => Some(project.clone()),
3010×
            _ => None,
302
        };
3034×
        let mut relations = (Vec::<String>::new(), Vec::<String>::new());
3044×
        if let Some(p) = &project {
3054×
            relations.0 = sqlx::query_scalar("SELECT depends_on FROM project_relations WHERE project = ?")
3064×
                .bind(p)
3074×
                .fetch_all(d.ledger.pool())
3084×
                .await
3094×
                .unwrap_or_default();
3104×
            relations.1 = sqlx::query_scalar("SELECT project FROM project_relations WHERE depends_on = ?")
3114×
                .bind(p)
3124×
                .fetch_all(d.ledger.pool())
3134×
                .await
3144×
                .unwrap_or_default();
3150×
        }
3164×
        let items = conv::inbox(&d.ledger, &ctx.actor(), Some(&ctx.author())).await.unwrap_or_default();
3174×
        let others = agents(ctx, json!({})).await;
3184×
        let _ = d.ledger.append_one(ctx.event("rules.read", json!({ "version": rules.version }))).await;
3194×
        let mut text = format!("You are {me}.");
3204×
        if let Some(p) = &project {
3214×
            text.push_str(&format!(" Your project is {p}; your channel is #{p}."));
3224×
            if !relations.0.is_empty() {
3230×
                text.push_str(&format!(" It pins {}.", relations.0.join(", ")));
3244×
            }
3254×
            if !relations.1.is_empty() {
3260×
                text.push_str(&format!(" {} pin it.", relations.1.join(", ")));
3274×
            }
3280×
        }
3294×
        if let Some(sess) = &ctx.session {
3304×
            text.push_str(&format!(" Session {sess}."));
3314×
        }
3324×
        text.push_str(&format!("\n\nYour brief (version {}):\n", rules.version));
3334×
        text.push_str(if rules.text.is_empty() { "(none set yet)\n" } else { &rules.text });
3344×
        text.push_str(&format!("\nInbox: {} unread.\n", items.len()));
3354×
        for i in &items {
3360×
            text.push_str(&format!("- {} from {} in {} ({}): {}\n", i.kind, label(d, &i.from).await, i.channel, i.id, i.summary));
337
        }
3384×
        text.push_str(&format!("\nAgents:\n{}\n", others.text));
3394×
        text.push_str("\nWhen you have nothing to do, end your turn: Spectra starts you again when there is work.");
3404×
        Output::ok(
3414×
            text,
3424×
            json!({ "you": me, "project": project, "session": ctx.session, "rules": { "version": rules.version, "text": rules.text },
3434×
                "inbox": items, "agents": others.structured.and_then(|s| s.get("agents").cloned()), "pins": relations.0, "pinned_by": relations.1 }),
344
        )
3454×
    })
3464×
}
347
3487×
fn rules(ctx: &Ctx, _args: Json) -> BoxFuture<'_, Output> {
3497×
    Box::pin(async move {
3507×
        match spectra_core::rules::compile(ctx.daemon.ledger.pool(), project_of(&ctx.caller.principal)).await {
3517×
            Ok(r) => {
3527×
                let _ = ctx.daemon.ledger.append_one(ctx.event("rules.read", json!({ "version": r.version }))).await;
3537×
                let body = if r.text.is_empty() { "(no brief set yet)".to_string() } else { r.text.clone() };
3547×
                Output::ok(format!("Your brief, version {}:\n\n{body}", r.version), json!({ "version": r.version, "text": r.text }))
355
            }
3560×
            Err(e) => fail(e),
357
        }
3587×
    })
3597×
}
360
3617×
fn agents(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
3627×
    Box::pin(async move {
3637×
        let project = s(&args, "project");
3647×
        let rows: Vec<AgentRow> = match sqlx::query_as(
3657×
            "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 \
3667×
             WHERE a.disabled_ms IS NULL AND (? IS NULL OR a.project = ?) ORDER BY a.name",
367
        )
3687×
        .bind(project)
3697×
        .bind(project)
3707×
        .fetch_all(ctx.daemon.ledger.pool())
3717×
        .await
372
        {
3737×
            Ok(r) => r,
3740×
            Err(e) => return fail(e.into()),
375
        };
376
        // Presence as the daemon sees it now; the table is its recorded history.
3777×
        let rows: Vec<_> = {
3787×
            let p = ctx.daemon.presence.lock().await;
3797×
            let now = now_ms();
3807×
            rows.into_iter()
38111×
                .map(|(name, owner, kind, project, _, status)| {
38211×
                    let live = p.state(&name, now).to_string();
38311×
                    (name, owner, kind, project, Some(live), status)
38411×
                })
3857×
                .collect()
386
        };
3877×
        let list: Vec<Json> = rows
3887×
            .iter()
38911×
            .map(|(name, owner, kind, project, presence, status)| {
39011×
                json!({ "name": name, "owner": owner, "kind": kind, "project": project, "presence": presence.clone().unwrap_or_else(|| "offline".into()), "status": status })
39111×
            })
3927×
            .collect();
3937×
        let text = if rows.is_empty() {
3940×
            "No agents.".to_string()
395
        } else {
3967×
            rows.iter()
39711×
                .map(|(name, owner, kind, project, presence, status)| {
39811×
                    let st = status.as_deref().map(|s| format!(" — \"{s}\"")).unwrap_or_default();
39911×
                    format!("- {name} · {owner}'s agent ({kind}, #{project}): {}{st}", presence.as_deref().unwrap_or("offline"))
40011×
                })
4017×
                .collect::<Vec<_>>()
4027×
                .join("\n")
403
        };
4047×
        Output::ok(text, json!({ "agents": list }))
4057×
    })
4067×
}
407
408
/// An agent as the `agents` tool reads it: name, owner, kind, project,
409
/// presence and status.
410
type AgentRow = (String, String, String, String, Option<String>, Option<String>);
411
4121×
fn status(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
4131×
    Box::pin(async move {
4141×
        let Principal::Agent { name, .. } = &ctx.caller.principal else { return Output::err("status is for agents") };
4151×
        let state = s(&args, "state").unwrap_or("working");
4161×
        let text = s(&args, "text");
4171×
        if text.is_some_and(|t| t.len() > 500) {
4180×
            return Output::err("a status is at most 500 characters");
4191×
        }
4201×
        let e = ctx.event("agent.status", json!({ "state": state, "text": text, "task": s(&args, "task") })).subject(name.clone());
4211×
        if let Err(e) = ctx.daemon.ledger.append_one(e).await {
4220×
            return fail(e);
4231×
        }
424
        {
4251×
            let mut p = ctx.daemon.presence.lock().await;
4261×
            p.status(name, (state != "idle").then(now_ms));
4271×
            p.blocked(name, state == "blocked");
428
        }
4291×
        Output::ok(
4301×
            format!("Status set: {state}{}", text.map(|t| format!(" — {t}")).unwrap_or_default()),
4311×
            json!({ "state": state, "text": text }),
432
        )
4331×
    })
4341×
}
435
4367×
fn say(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
4377×
    Box::pin(async move {
4387×
        let Some(body) = s(&args, "body") else { return Output::err("say needs a body") };
4397×
        let mut channel = s(&args, "channel").map(String::from);
4407×
        if let Some(to) = s(&args, "to") {
4410×
            match conv::ensure_direct(&ctx.daemon.ledger, &ctx.caller.principal, to).await {
4420×
                Ok(c) => channel = Some(c.name),
4430×
                Err(e) => return fail(e),
444
            }
4457×
        }
4467×
        let post = Post {
4477×
            channel,
4487×
            body: body.into(),
4497×
            reply_to: s(&args, "reply_to").map(String::from),
4507×
            kind: s(&args, "kind").map(String::from),
4517×
            ..Default::default()
4527×
        };
4537×
        match conv::post(&ctx.daemon.ledger, &ctx.author(), post, now_ms()).await {
4546×
            Ok(Posted::Message { id, channel }) => {
4556×
                Output::ok(format!("Posted {id} in {channel}."), json!({ "id": id, "channel": channel }))
456
            }
4570×
            Ok(Posted::Duplicate { id }) => {
4580×
                Output::ok(format!("Not posted: you said exactly this in the last minute ({id})."), json!({ "duplicate_of": id }))
459
            }
4601×
            Err(e) => fail(e),
461
        }
4627×
    })
4637×
}
464
4651×
fn read(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
4661×
    Box::pin(async move {
4671×
        let limit = args.get("limit").and_then(Json::as_u64).unwrap_or(50) as u32;
4681×
        match conv::read(&ctx.daemon.ledger, &ctx.caller.principal, s(&args, "channel"), s(&args, "thread"), s(&args, "since"), limit).await
469
        {
4701×
            Ok((ch, msgs)) => {
4711×
                let mut text = format!("{} message(s) in {}:\n\n", msgs.len(), ch.display());
4721×
                for m in &msgs {
4731×
                    text.push_str(&delimited(&ctx.daemon, m, &ch.display()).await);
4741×
                    text.push_str("\n\n");
475
                }
4761×
                Output::ok(text.trim_end().to_string(), json!({ "channel": ch.display(), "messages": msgs }))
477
            }
4780×
            Err(e) => fail(e),
479
        }
4801×
    })
4811×
}
482
4835×
fn inbox(ctx: &Ctx, _args: Json) -> BoxFuture<'_, Output> {
4845×
    Box::pin(async move {
4855×
        match conv::inbox(&ctx.daemon.ledger, &ctx.actor(), Some(&ctx.author())).await {
4865×
            Ok(items) => inbox_output(ctx, items).await,
4870×
            Err(e) => fail(e),
488
        }
4895×
    })
4905×
}
491
4927×
async fn inbox_output(ctx: &Ctx, items: Vec<conv::Item>) -> Output {
4937×
    if items.is_empty() {
4945×
        return Output::ok("Nothing unread.", json!({ "items": [] }));
4952×
    }
4962×
    let mut text = format!("{} unread:\n\n", items.len());
4972×
    for i in &items {
4982×
        text.push_str(&format!(
499
            "[{} from {} in {}, id {}]\n{}\n[end]\nAct on it: {}\n\n",
500
            i.kind,
5012×
            label(&ctx.daemon, &i.from).await,
502
            i.channel,
503
            i.id,
504
            i.summary,
505
            i.action_hint
506
        ));
507
    }
5082×
    Output::ok(text.trim_end().to_string(), json!({ "items": items }))
5097×
}
510
5116×
fn wait(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
5126×
    Box::pin(async move {
5136×
        let actor = ctx.actor();
5146×
        let now = now_ms();
515
        {
5166×
            let guard = ctx.daemon.idle_waits.lock().await;
5176×
            if let Some((_, until)) = guard.get(&actor) {
5183×
                if *until > now {
5191×
                    return Output::err("Nothing is waiting for you. End your turn; Spectra starts you again when there is work.");
5202×
                }
5213×
            }
522
        }
5235×
        let timeout = args.get("timeout_s").and_then(Json::as_u64).unwrap_or(30).clamp(1, 50);
5245×
        let name = actor.strip_prefix("agent:").unwrap_or(&actor).to_string();
525
        // Subscribe before looking, so nothing is missed between the two.
5265×
        let mut rx = ctx.daemon.ledger.subscribe();
5275×
        let ledger = &ctx.daemon.ledger;
5285×
        ctx.daemon.presence.lock().await.wait_started(&name);
5295×
        let found = async {
530
            loop {
5318×
                if conv::unread(ledger, &actor).await.unwrap_or(0) > 0 {
5322×
                    return true;
5336×
                }
5346×
                match rx.recv().await {
5353×
                    Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
5360×
                    Err(_) => return false,
537
                }
538
            }
5392×
        };
5405×
        let got = tokio::time::timeout(Duration::from_secs(timeout), found).await.unwrap_or(false);
5415×
        ctx.daemon.presence.lock().await.wait_ended(&name);
5425×
        let mut guard = ctx.daemon.idle_waits.lock().await;
5435×
        if got {
5442×
            guard.remove(&actor);
5452×
            drop(guard);
5462×
            return match conv::inbox(ledger, &actor, Some(&ctx.author())).await {
5472×
                Ok(items) => inbox_output(ctx, items).await,
5480×
                Err(e) => fail(e),
549
            };
5503×
        }
5513×
        let entry = guard.entry(actor).or_insert((0, 0));
5523×
        entry.0 += 1;
5533×
        if entry.0 >= IDLE_WAITS {
5541×
            entry.1 = now_ms() + IDLE_COOLDOWN_MS;
5551×
            entry.0 = 0;
5562×
        }
5573×
        Output { text: "Nothing new.".into(), structured: Some(json!({ "items": [] })), is_error: false, quiet: true }
5586×
    })
5596×
}
560
5611×
fn ack(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
5621×
    Box::pin(async move {
5631×
        let ids: Vec<String> = args
5641×
            .get("ids")
5651×
            .and_then(Json::as_array)
5661×
            .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
5671×
            .unwrap_or_default();
5681×
        if ids.is_empty() {
5690×
            return Output::err("ack needs ids");
5701×
        }
5711×
        match conv::ack(&ctx.daemon.ledger, &ctx.author(), &ids, s(&args, "reaction")).await {
5721×
            Ok(n) => Output::ok(format!("Marked {n} read."), json!({ "read": n })),
5730×
            Err(e) => fail(e),
574
        }
5751×
    })
5761×
}
577
5781×
fn search(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
5791×
    Box::pin(async move {
5801×
        let Some(q) = s(&args, "query") else { return Output::err("search needs a query") };
5811×
        let since = s(&args, "since").and_then(|v| parse_since(v, now_ms()));
5821×
        let limit = args.get("limit").and_then(Json::as_u64).unwrap_or(20) as u32;
5831×
        match conv::search(&ctx.daemon.ledger, &ctx.caller.principal, q, s(&args, "channel"), s(&args, "author"), since, limit).await {
5841×
            Ok(hits) => {
5851×
                let mut text = format!("{} hit(s):\n", hits.len());
5861×
                for h in &hits {
5871×
                    text.push_str(&format!(
588
                        "- [{} in {} at {}, id {}] {}\n",
5891×
                        label(&ctx.daemon, &h.author).await,
590
                        h.channel,
5911×
                        spectra_core::ids::format_time(h.at_ms as u64),
592
                        h.id,
593
                        h.snippet
594
                    ));
595
                }
5961×
                Output::ok(text.trim_end().to_string(), json!({ "hits": hits }))
597
            }
5980×
            Err(e) => fail(e),
599
        }
6001×
    })
6011×
}
602
603
/// `7d`, `3h`, `30m`, or milliseconds since the epoch.
6041×
pub fn parse_since(v: &str, now: u64) -> Option<i64> {
6051×
    let (n, unit) = v.split_at(v.len().saturating_sub(1));
6061×
    let ms = match unit {
6071×
        "d" => 86_400_000,
6080×
        "h" => 3_600_000,
6090×
        "m" => 60_000,
6100×
        _ => return v.parse().ok(),
611
    };
6121×
    n.parse::<u64>().ok().map(|n| now.saturating_sub(n * ms) as i64)
6131×
}
614
6152×
fn redact(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
6162×
    Box::pin(async move {
6172×
        let (Some(id), Some(reason)) = (s(&args, "id"), s(&args, "reason")) else { return Output::err("redact needs id and reason") };
6182×
        match conv::redact(&ctx.daemon.ledger, &ctx.author(), id, reason).await {
6191×
            Ok(()) => Output::ok(format!("Redacted {id}."), json!({ "id": id })),
6201×
            Err(e) => fail(e),
621
        }
6222×
    })
6232×
}
624
625
/// The project whose brief applies to `p`: an agent's own; none for the owner.
626128×
pub fn project_of(p: &Principal) -> Option<&str> {
627128×
    match p {
628128×
        Principal::Agent { project, .. } => Some(project.as_str()),
6290×
        _ => None,
630
    }
631128×
}