OXYGENHUB
◆ v0.1.0 spectra / crates / spectra-mcp / src / tools.rs
24.4 KiB · Rust · last changed 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) 2 days ago source blame history raw
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 {
35 
    fn author(&self) -> Author<'static> {
36 
        Author { principal: self.caller.principal.clone(), signer: Signer::Request(self.caller.auth.clone()) }
37 
    }
38 
    fn actor(&self) -> String {
39 
        self.caller.principal.actor()
40 
    }
41 
    fn event(&self, kind: &str, payload: Json) -> NewEvent {
42 
        let mut e = NewEvent::system(kind, payload).by(self.actor(), self.caller.auth.clone());
43 
        e.session = self.session.clone();
44 
        e
45 
    }
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 {
60 
    pub fn ok(text: impl Into<String>, structured: Json) -> Output {
61 
        Output { text: text.into(), structured: Some(structured), ..Default::default() }
62 
    }
63 
    pub fn err(text: impl Into<String>) -> Output {
64 
        Output { text: text.into(), is_error: true, ..Default::default() }
65 
    }
66 
}
67
68 
fn fail(e: anyhow::Error) -> Output {
69 
    Output::err(format!("{e:#}"))
70 
}
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 {
87 
    pub fn describe(&self) -> Json {
88 
        json!({
89 
            "name": self.name,
90 
            "description": self.description,
91 
            "inputSchema": (self.schema)(),
92 
            "annotations": { "readOnlyHint": self.read_only, "destructiveHint": self.destructive },
93 
        })
94 
    }
95 
}
96
97 
/// Every tool, in the order `tools/list` gives them.
98 
pub fn all() -> &'static [Tool] {
99 
    TOOLS
100 
}
101
102 
pub fn find(name: &str) -> Option<&'static Tool> {
103 
    TOOLS.iter().find(|t| t.name == name)
104 
}
105
106 
/// Run a tool for the caller: the idle guard, the rules line, and the record.
107 
pub async fn invoke(ctx: &Ctx, tool: &Tool, args: Json) -> Output {
108 
    let started = std::time::Instant::now();
109 
    let mut out = (tool.run)(ctx, args.clone()).await;
110 
    let Principal::Agent { name, .. } = &ctx.caller.principal else { return out };
111 
    if tool.name != "rules" && tool.name != "brief" {
112 
        // With no house rules there is nothing to read, so nothing to say.
113 
        if let Some(rules) = spectra_core::rules::compile(&ctx.daemon.data).ok().filter(|r| !r.text.is_empty()) {
114 
            let read = spectra_core::rules::last_read(ctx.daemon.ledger.pool(), name).await.ok().flatten();
115 
            if read.as_deref() != Some(rules.version.as_str()) {
116 
                let from = read.unwrap_or_else(|| "none".into());
117 
                out.text = format!("{}\n\n{}", spectra_core::rules::changed_line(&from, &rules.version), out.text);
118 
            }
119 
        }
120 
    }
121 
    if out.quiet {
122 
        return out;
123 
    }
124 
    let data = &ctx.daemon.data;
125 
    let args_hash = spectra_core::attachments::put(data, args.to_string().as_bytes()).unwrap_or_default();
126 
    let result_hash = spectra_core::attachments::put(data, out.text.as_bytes()).unwrap_or_default();
127 
    let e = ctx.event(
128 
        "tool.called",
129 
        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 }),
130 
    );
131 
    let _ = ctx.daemon.ledger.append_one(e).await;
132 
    out
133 
}
134
135 
fn obj(props: Json, required: &[&str]) -> Json {
136 
    json!({ "type": "object", "properties": props, "required": required, "additionalProperties": false })
137 
}
138
139 
static TOOLS: &[Tool] = &[
140 
    Tool {
141 
        name: "brief",
142 
        description: "Call first: who you are, your project, the house rules, your unread inbox, and who else is here.",
143 
        schema: || obj(json!({ "new_session": { "type": "boolean" } }), &[]),
144 
        read_only: true,
145 
        destructive: false,
146 
        long: false,
147 
        run: brief,
148 
    },
149 
    Tool {
150 
        name: "rules",
151 
        description: "The house rules and their version, in full.",
152 
        schema: || obj(json!({}), &[]),
153 
        read_only: true,
154 
        destructive: false,
155 
        long: false,
156 
        run: rules,
157 
    },
158 
    Tool {
159 
        name: "agents",
160 
        description: "Every agent with its owner, kind, project, presence and status.",
161 
        schema: || obj(json!({ "project": { "type": "string" } }), &[]),
162 
        read_only: true,
163 
        destructive: false,
164 
        long: false,
165 
        run: agents,
166 
    },
167 
    Tool {
168 
        name: "status",
169 
        description: "Say what you are doing, so others can see.",
170 
        schema: || obj(json!({ "state": { "enum": ["working", "idle", "blocked"] }, "text": { "type": "string" }, "task": { "type": "string" } }), &[]),
171 
        read_only: false,
172 
        destructive: false,
173 
        long: false,
174 
        run: status,
175 
    },
176 
    Tool {
177 
        name: "say",
178 
        description: "Post a message (default: your project's channel). @name mentions; reply_to answers in a thread.",
179 
        schema: || {
180 
            obj(
181 
                json!({ "body": { "type": "string" }, "channel": { "type": "string" }, "reply_to": { "type": "string" },
182 
                    "kind": { "enum": ["chat", "request", "answer", "handoff", "status", "decision"] }, "to": { "type": "string" } }),
183 
                &["body"],
184 
            )
185 
        },
186 
        read_only: false,
187 
        destructive: false,
188 
        long: false,
189 
        run: say,
190 
    },
191 
    Tool {
192 
        name: "read",
193 
        description: "Messages of a channel (default: your project's) or a whole thread; marks nothing read.",
194 
        schema: || obj(json!({ "channel": { "type": "string" }, "thread": { "type": "string" }, "since": { "type": "string" }, "limit": { "type": "integer" } }), &[]),
195 
        read_only: true,
196 
        destructive: false,
197 
        long: false,
198 
        run: read,
199 
    },
200 
    Tool {
201 
        name: "inbox",
202 
        description: "Your unread mentions and direct messages.",
203 
        schema: || obj(json!({}), &[]),
204 
        read_only: true,
205 
        destructive: false,
206 
        long: false,
207 
        run: inbox,
208 
    },
209 
    Tool {
210 
        name: "wait",
211 
        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.",
212 
        schema: || obj(json!({ "timeout_s": { "type": "integer", "minimum": 1, "maximum": 50 } }), &["timeout_s"]),
213 
        read_only: true,
214 
        destructive: false,
215 
        long: true,
216 
        run: wait,
217 
    },
218 
    Tool {
219 
        name: "ack",
220 
        description: "Mark inbox items read, with an optional reaction (ack, done, eyes, no).",
221 
        schema: || obj(json!({ "ids": { "type": "array", "items": { "type": "string" } }, "reaction": { "enum": ["ack", "done", "eyes", "no"] } }), &["ids"]),
222 
        read_only: false,
223 
        destructive: false,
224 
        long: false,
225 
        run: ack,
226 
    },
227 
    Tool {
228 
        name: "search",
229 
        description: "Search history: words, \"phrases\", prefix*, AND/OR/NOT; filter by channel, author, since (7d, 3h).",
230 
        schema: || {
231 
            obj(json!({ "query": { "type": "string" }, "channel": { "type": "string" }, "author": { "type": "string" }, "since": { "type": "string" }, "limit": { "type": "integer" } }), &["query"])
232 
        },
233 
        read_only: true,
234 
        destructive: false,
235 
        long: false,
236 
        run: search,
237 
    },
238 
    Tool {
239 
        name: "redact",
240 
        description: "Redact your own message, with the reason.",
241 
        schema: || obj(json!({ "id": { "type": "string" }, "reason": { "type": "string" } }), &["id", "reason"]),
242 
        read_only: false,
243 
        destructive: true,
244 
        long: false,
245 
        run: redact,
246 
    },
247 
];
248
249 
fn s<'a>(args: &'a Json, k: &str) -> Option<&'a str> {
250 
    args.get(k).and_then(Json::as_str).filter(|v| !v.is_empty())
251 
}
252
253 
/// How a principal is named to a model: `claude-oxygen (Andy McCall's agent, claude-code)`.
254 
async fn label(d: &Daemon, actor: &str) -> String {
255 
    if actor == "human" {
256 
        let name: Option<String> = sqlx::query_scalar("SELECT name FROM identities WHERE role = 'owner' AND current = 1")
257 
            .fetch_optional(d.ledger.pool())
258 
            .await
259 
            .ok()
260 
            .flatten();
261 
        return format!("{} (the human)", name.unwrap_or_else(|| "the owner".into()));
262 
    }
263 
    if let Some(a) = actor.strip_prefix("agent:") {
264 
        let row: Option<(String, String)> = sqlx::query_as("SELECT owner_name, kind FROM agents WHERE name = ?")
265 
            .bind(a)
266 
            .fetch_optional(d.ledger.pool())
267 
            .await
268 
            .ok()
269 
            .flatten();
270 
        if let Some((owner, kind)) = row {
271 
            return format!("{a} ({owner}'s agent, {kind})");
272 
        }
273 
        return a.to_string();
274 
    }
275 
    "Spectra".into()
276 
}
277
278 
/// A message as a model reads it: delimited with its author (spec §20.3).
279 
async fn delimited(d: &Daemon, m: &conv::Message, channel: &str) -> String {
280 
    let at = spectra_core::ids::format_time(m.at_ms as u64);
281 
    let thread = m.reply_to.as_deref().map(|r| format!(", reply in thread {r}")).unwrap_or_default();
282 
    let replies = if m.replies > 0 { format!(" ({} replies)", m.replies) } else { String::new() };
283 
    format!("[from {} at {at} in {channel}, id {}{thread}{replies}]\n{}\n[end]", label(d, &m.author).await, m.id, m.body)
284 
}
285
286 
fn brief(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
287 
    let _ = args;
288 
    Box::pin(async move {
289 
        let d = &ctx.daemon;
290 
        let rules = match spectra_core::rules::compile(&d.data) {
291 
            Ok(r) => r,
292 
            Err(e) => return fail(e),
293 
        };
294 
        let me = ctx.caller.principal.display();
295 
        let project = match &ctx.caller.principal {
296 
            Principal::Agent { project, .. } => Some(project.clone()),
297 
            _ => None,
298 
        };
299 
        let mut relations = (Vec::<String>::new(), Vec::<String>::new());
300 
        if let Some(p) = &project {
301 
            relations.0 = sqlx::query_scalar("SELECT depends_on FROM project_relations WHERE project = ?")
302 
                .bind(p)
303 
                .fetch_all(d.ledger.pool())
304 
                .await
305 
                .unwrap_or_default();
306 
            relations.1 = sqlx::query_scalar("SELECT project FROM project_relations WHERE depends_on = ?")
307 
                .bind(p)
308 
                .fetch_all(d.ledger.pool())
309 
                .await
310 
                .unwrap_or_default();
311 
        }
312 
        let items = conv::inbox(&d.ledger, &ctx.actor(), Some(&ctx.author())).await.unwrap_or_default();
313 
        let others = agents(ctx, json!({})).await;
314 
        let _ = d.ledger.append_one(ctx.event("rules.read", json!({ "version": rules.version }))).await;
315 
        let mut text = format!("You are {me}.");
316 
        if let Some(p) = &project {
317 
            text.push_str(&format!(" Your project is {p}; your channel is #{p}."));
318 
            if !relations.0.is_empty() {
319 
                text.push_str(&format!(" It pins {}.", relations.0.join(", ")));
320 
            }
321 
            if !relations.1.is_empty() {
322 
                text.push_str(&format!(" {} pin it.", relations.1.join(", ")));
323 
            }
324 
        }
325 
        if let Some(sess) = &ctx.session {
326 
            text.push_str(&format!(" Session {sess}."));
327 
        }
328 
        text.push_str(&format!("\n\nHouse rules (version {}):\n", rules.version));
329 
        text.push_str(if rules.text.is_empty() { "(none yet)\n" } else { &rules.text });
330 
        text.push_str(&format!("\nInbox: {} unread.\n", items.len()));
331 
        for i in &items {
332 
            text.push_str(&format!("- {} from {} in {} ({}): {}\n", i.kind, label(d, &i.from).await, i.channel, i.id, i.summary));
333 
        }
334 
        text.push_str(&format!("\nAgents:\n{}\n", others.text));
335 
        text.push_str("\nWhen you have nothing to do, end your turn: Spectra starts you again when there is work.");
336 
        Output::ok(
337 
            text,
338 
            json!({ "you": me, "project": project, "session": ctx.session, "rules": { "version": rules.version, "text": rules.text },
339 
                "inbox": items, "agents": others.structured.and_then(|s| s.get("agents").cloned()), "pins": relations.0, "pinned_by": relations.1 }),
340 
        )
341 
    })
342 
}
343
344 
fn rules(ctx: &Ctx, _args: Json) -> BoxFuture<'_, Output> {
345 
    Box::pin(async move {
346 
        match spectra_core::rules::compile(&ctx.daemon.data) {
347 
            Ok(r) => {
348 
                let _ = ctx.daemon.ledger.append_one(ctx.event("rules.read", json!({ "version": r.version }))).await;
349 
                let body = if r.text.is_empty() { "(no house rules yet)".to_string() } else { r.text.clone() };
350 
                Output::ok(format!("House rules, version {}:\n\n{body}", r.version), json!({ "version": r.version, "text": r.text }))
351 
            }
352 
            Err(e) => fail(e),
353 
        }
354 
    })
355 
}
356
357 
fn agents(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
358 
    Box::pin(async move {
359 
        let project = s(&args, "project");
360 
        let rows: Vec<AgentRow> = match sqlx::query_as(
361 
            "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 \
362 
             WHERE a.disabled_ms IS NULL AND (? IS NULL OR a.project = ?) ORDER BY a.name",
363 
        )
364 
        .bind(project)
365 
        .bind(project)
366 
        .fetch_all(ctx.daemon.ledger.pool())
367 
        .await
368 
        {
369 
            Ok(r) => r,
370 
            Err(e) => return fail(e.into()),
371 
        };
372 
        // Presence as the daemon sees it now; the table is its recorded history.
373 
        let rows: Vec<_> = {
374 
            let p = ctx.daemon.presence.lock().await;
375 
            let now = now_ms();
376 
            rows.into_iter()
377 
                .map(|(name, owner, kind, project, _, status)| {
378 
                    let live = p.state(&name, now).to_string();
379 
                    (name, owner, kind, project, Some(live), status)
380 
                })
381 
                .collect()
382 
        };
383 
        let list: Vec<Json> = rows
384 
            .iter()
385 
            .map(|(name, owner, kind, project, presence, status)| {
386 
                json!({ "name": name, "owner": owner, "kind": kind, "project": project, "presence": presence.clone().unwrap_or_else(|| "offline".into()), "status": status })
387 
            })
388 
            .collect();
389 
        let text = if rows.is_empty() {
390 
            "No agents.".to_string()
391 
        } else {
392 
            rows.iter()
393 
                .map(|(name, owner, kind, project, presence, status)| {
394 
                    let st = status.as_deref().map(|s| format!(" — \"{s}\"")).unwrap_or_default();
395 
                    format!("- {name} · {owner}'s agent ({kind}, #{project}): {}{st}", presence.as_deref().unwrap_or("offline"))
396 
                })
397 
                .collect::<Vec<_>>()
398 
                .join("\n")
399 
        };
400 
        Output::ok(text, json!({ "agents": list }))
401 
    })
402 
}
403
404 
/// An agent as the `agents` tool reads it: name, owner, kind, project,
405 
/// presence and status.
406 
type AgentRow = (String, String, String, String, Option<String>, Option<String>);
407
408 
fn status(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
409 
    Box::pin(async move {
410 
        let Principal::Agent { name, .. } = &ctx.caller.principal else { return Output::err("status is for agents") };
411 
        let state = s(&args, "state").unwrap_or("working");
412 
        let text = s(&args, "text");
413 
        if text.is_some_and(|t| t.len() > 500) {
414 
            return Output::err("a status is at most 500 characters");
415 
        }
416 
        let e = ctx.event("agent.status", json!({ "state": state, "text": text, "task": s(&args, "task") })).subject(name.clone());
417 
        if let Err(e) = ctx.daemon.ledger.append_one(e).await {
418 
            return fail(e);
419 
        }
420 
        {
421 
            let mut p = ctx.daemon.presence.lock().await;
422 
            p.status(name, (state != "idle").then(now_ms));
423 
            p.blocked(name, state == "blocked");
424 
        }
425 
        Output::ok(
426 
            format!("Status set: {state}{}", text.map(|t| format!(" — {t}")).unwrap_or_default()),
427 
            json!({ "state": state, "text": text }),
428 
        )
429 
    })
430 
}
431
432 
fn say(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
433 
    Box::pin(async move {
434 
        let Some(body) = s(&args, "body") else { return Output::err("say needs a body") };
435 
        let mut channel = s(&args, "channel").map(String::from);
436 
        if let Some(to) = s(&args, "to") {
437 
            match conv::ensure_direct(&ctx.daemon.ledger, &ctx.caller.principal, to).await {
438 
                Ok(c) => channel = Some(c.name),
439 
                Err(e) => return fail(e),
440 
            }
441 
        }
442 
        let post = Post {
443 
            channel,
444 
            body: body.into(),
445 
            reply_to: s(&args, "reply_to").map(String::from),
446 
            kind: s(&args, "kind").map(String::from),
447 
            ..Default::default()
448 
        };
449 
        match conv::post(&ctx.daemon.ledger, &ctx.author(), post, now_ms()).await {
450 
            Ok(Posted::Message { id, channel }) => {
451 
                Output::ok(format!("Posted {id} in {channel}."), json!({ "id": id, "channel": channel }))
452 
            }
453 
            Ok(Posted::Duplicate { id }) => {
454 
                Output::ok(format!("Not posted: you said exactly this in the last minute ({id})."), json!({ "duplicate_of": id }))
455 
            }
456 
            Err(e) => fail(e),
457 
        }
458 
    })
459 
}
460
461 
fn read(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
462 
    Box::pin(async move {
463 
        let limit = args.get("limit").and_then(Json::as_u64).unwrap_or(50) as u32;
464 
        match conv::read(&ctx.daemon.ledger, &ctx.caller.principal, s(&args, "channel"), s(&args, "thread"), s(&args, "since"), limit).await
465 
        {
466 
            Ok((ch, msgs)) => {
467 
                let mut text = format!("{} message(s) in {}:\n\n", msgs.len(), ch.display());
468 
                for m in &msgs {
469 
                    text.push_str(&delimited(&ctx.daemon, m, &ch.display()).await);
470 
                    text.push_str("\n\n");
471 
                }
472 
                Output::ok(text.trim_end().to_string(), json!({ "channel": ch.display(), "messages": msgs }))
473 
            }
474 
            Err(e) => fail(e),
475 
        }
476 
    })
477 
}
478
479 
fn inbox(ctx: &Ctx, _args: Json) -> BoxFuture<'_, Output> {
480 
    Box::pin(async move {
481 
        match conv::inbox(&ctx.daemon.ledger, &ctx.actor(), Some(&ctx.author())).await {
482 
            Ok(items) => inbox_output(ctx, items).await,
483 
            Err(e) => fail(e),
484 
        }
485 
    })
486 
}
487
488 
async fn inbox_output(ctx: &Ctx, items: Vec<conv::Item>) -> Output {
489 
    if items.is_empty() {
490 
        return Output::ok("Nothing unread.", json!({ "items": [] }));
491 
    }
492 
    let mut text = format!("{} unread:\n\n", items.len());
493 
    for i in &items {
494 
        text.push_str(&format!(
495 
            "[{} from {} in {}, id {}]\n{}\n[end]\nAct on it: {}\n\n",
496 
            i.kind,
497 
            label(&ctx.daemon, &i.from).await,
498 
            i.channel,
499 
            i.id,
500 
            i.summary,
501 
            i.action_hint
502 
        ));
503 
    }
504 
    Output::ok(text.trim_end().to_string(), json!({ "items": items }))
505 
}
506
507 
fn wait(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
508 
    Box::pin(async move {
509 
        let actor = ctx.actor();
510 
        let now = now_ms();
511 
        {
512 
            let guard = ctx.daemon.idle_waits.lock().await;
513 
            if let Some((_, until)) = guard.get(&actor) {
514 
                if *until > now {
515 
                    return Output::err("Nothing is waiting for you. End your turn; Spectra starts you again when there is work.");
516 
                }
517 
            }
518 
        }
519 
        let timeout = args.get("timeout_s").and_then(Json::as_u64).unwrap_or(30).clamp(1, 50);
520 
        let name = actor.strip_prefix("agent:").unwrap_or(&actor).to_string();
521 
        // Subscribe before looking, so nothing is missed between the two.
522 
        let mut rx = ctx.daemon.ledger.subscribe();
523 
        let ledger = &ctx.daemon.ledger;
524 
        ctx.daemon.presence.lock().await.wait_started(&name);
525 
        let found = async {
526 
            loop {
527 
                if conv::unread(ledger, &actor).await.unwrap_or(0) > 0 {
528 
                    return true;
529 
                }
530 
                match rx.recv().await {
531 
                    Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
532 
                    Err(_) => return false,
533 
                }
534 
            }
535 
        };
536 
        let got = tokio::time::timeout(Duration::from_secs(timeout), found).await.unwrap_or(false);
537 
        ctx.daemon.presence.lock().await.wait_ended(&name);
538 
        let mut guard = ctx.daemon.idle_waits.lock().await;
539 
        if got {
540 
            guard.remove(&actor);
541 
            drop(guard);
542 
            return match conv::inbox(ledger, &actor, Some(&ctx.author())).await {
543 
                Ok(items) => inbox_output(ctx, items).await,
544 
                Err(e) => fail(e),
545 
            };
546 
        }
547 
        let entry = guard.entry(actor).or_insert((0, 0));
548 
        entry.0 += 1;
549 
        if entry.0 >= IDLE_WAITS {
550 
            entry.1 = now_ms() + IDLE_COOLDOWN_MS;
551 
            entry.0 = 0;
552 
        }
553 
        Output { text: "Nothing new.".into(), structured: Some(json!({ "items": [] })), is_error: false, quiet: true }
554 
    })
555 
}
556
557 
fn ack(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
558 
    Box::pin(async move {
559 
        let ids: Vec<String> = args
560 
            .get("ids")
561 
            .and_then(Json::as_array)
562 
            .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
563 
            .unwrap_or_default();
564 
        if ids.is_empty() {
565 
            return Output::err("ack needs ids");
566 
        }
567 
        match conv::ack(&ctx.daemon.ledger, &ctx.author(), &ids, s(&args, "reaction")).await {
568 
            Ok(n) => Output::ok(format!("Marked {n} read."), json!({ "read": n })),
569 
            Err(e) => fail(e),
570 
        }
571 
    })
572 
}
573
574 
fn search(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
575 
    Box::pin(async move {
576 
        let Some(q) = s(&args, "query") else { return Output::err("search needs a query") };
577 
        let since = s(&args, "since").and_then(|v| parse_since(v, now_ms()));
578 
        let limit = args.get("limit").and_then(Json::as_u64).unwrap_or(20) as u32;
579 
        match conv::search(&ctx.daemon.ledger, &ctx.caller.principal, q, s(&args, "channel"), s(&args, "author"), since, limit).await {
580 
            Ok(hits) => {
581 
                let mut text = format!("{} hit(s):\n", hits.len());
582 
                for h in &hits {
583 
                    text.push_str(&format!(
584 
                        "- [{} in {} at {}, id {}] {}\n",
585 
                        label(&ctx.daemon, &h.author).await,
586 
                        h.channel,
587 
                        spectra_core::ids::format_time(h.at_ms as u64),
588 
                        h.id,
589 
                        h.snippet
590 
                    ));
591 
                }
592 
                Output::ok(text.trim_end().to_string(), json!({ "hits": hits }))
593 
            }
594 
            Err(e) => fail(e),
595 
        }
596 
    })
597 
}
598
599 
/// `7d`, `3h`, `30m`, or milliseconds since the epoch.
600 
pub fn parse_since(v: &str, now: u64) -> Option<i64> {
601 
    let (n, unit) = v.split_at(v.len().saturating_sub(1));
602 
    let ms = match unit {
603 
        "d" => 86_400_000,
604 
        "h" => 3_600_000,
605 
        "m" => 60_000,
606 
        _ => return v.parse().ok(),
607 
    };
608 
    n.parse::<u64>().ok().map(|n| now.saturating_sub(n * ms) as i64)
609 
}
610
611 
fn redact(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {
612 
    Box::pin(async move {
613 
        let (Some(id), Some(reason)) = (s(&args, "id"), s(&args, "reason")) else { return Output::err("redact needs id and reason") };
614 
        match conv::redact(&ctx.daemon.ledger, &ctx.author(), id, reason).await {
615 
            Ok(()) => Output::ok(format!("Redacted {id}."), json!({ "id": id })),
616 
            Err(e) => fail(e),
617 
        }
618 
    })
619 
}