conversation.rs 89.5%

496 of 554 lines run, 68 of 159 functions · run #7 on main · run missed
//! Conversation (spec §10): channels, messages, threads, reactions, each
//! principal's inbox, and search (§13.1). The MCP tools, the API and the
//! CLI all go through here, so the rules are the same whoever asks.
//!
//! A message is redacted by its author or the owner: Spectra stops showing
//! it everywhere, while the ledger keeps the original for audit until a
//! prune removes it (§9.1).

use crate::event::{Auth, Event, NewEvent};
use crate::keys::KeyPair;
use crate::ledger::Ledger;
use crate::principals::{signed_event, Principal};
use anyhow::{bail, Context, Result};
use serde::Serialize;
use serde_json::{json, Value as Json};
use sqlx::SqliteConnection;

/// The most a body or a payload may be (spec §10.3).
pub const MAX_BODY: usize = 64 * 1024;
/// A message identical to one its author posted in the same channel within
/// this long is dropped (spec §20.4).
pub const DUPLICATE_MS: u64 = 60_000;
/// The channels every Spectra has (spec §10.1).
pub const DEFAULT_CHANNELS: &[(&str, &str, &str)] = &[
    ("general", "shared", "Everyone: agents and the human"),
    ("humans", "shared", "Where agents ask the human"),
    ("system", "system", "Guard rails, schedules and hooks; read-only for agents"),
];
/// Message kinds (spec §10.3).
pub const KINDS: &[&str] = &["chat", "request", "answer", "handoff", "status", "decision", "system"];
pub const REACTIONS: &[&str] = &["ack", "done", "eyes", "no"];

/// Who is acting, and how their events are signed.
pub struct Author<'a> {
    pub principal: Principal,
    pub signer: Signer<'a>,
}

/// The CLI signs each event over its payload; the daemon carries the proof
/// of the request that caused it.
pub enum Signer<'a> {
    Cli(&'a KeyPair),
    Request(Auth),
    System,
}

impl Author<'_> {
    fn event(&self, kind: &str, payload: Json) -> NewEvent {
        let actor = self.principal.actor();
        match &self.signer {
            Signer::Cli(pair) => signed_event(pair, &actor, kind, payload),
            Signer::Request(auth) => NewEvent::system(kind, payload).by(actor, auth.clone()),
            Signer::System => NewEvent::system(kind, payload),
        }
    }
}

#[derive(Clone, Debug, Serialize, sqlx::FromRow)]
pub struct Channel {
    pub id: String,
    pub name: String,
    pub kind: String,
    pub project: Option<String>,
    pub purpose: Option<String>,
    pub members: String,
}

impl Channel {
    pub fn members(&self) -> Vec<String> {
        serde_json::from_str(&self.members).unwrap_or_default()
    }
    /// `#general`, `#oxygen`, or `@a+b` for a direct channel.
    pub fn display(&self) -> String {
        match self.name.strip_prefix("dm:") {
            Some(pair) => format!("@{pair}"),
            None => format!("#{}", self.name),
        }
    }
}

#[derive(Clone, Debug, Serialize, sqlx::FromRow)]
pub struct Message {
    pub id: String,
    pub channel: String,
    pub reply_to: Option<String>,
    pub author: String,
    pub kind: String,
    pub body: String,
    pub payload: Option<String>,
    pub mentions: String,
    pub refs: String,
    pub at_ms: i64,
    pub seq: i64,
    pub replies: i64,
    pub redacted: Option<String>,
}

/// What a post asks for.
#[derive(Clone, Debug, Default)]
pub struct Post {
    pub channel: Option<String>,
    pub body: String,
    pub reply_to: Option<String>,
    pub kind: Option<String>,
    pub payload: Option<Json>,
    pub mentions: Vec<String>,
    pub refs: Vec<String>,
}

/// The outcome of a post: the message, or why it was dropped.
#[derive(Clone, Debug, Serialize)]
pub enum Posted {
    Message { id: String, channel: String },
    Duplicate { id: String },
}

/// Make the default channels (at `spectra init`).
pub fn default_channel_events() -> Vec<NewEvent> {
    DEFAULT_CHANNELS.iter().map(|(name, kind, purpose)| channel_event(name, kind, None, Some(purpose), &[])).collect()
}

/// A project's channel, made when the project is added.
pub fn project_channel_event(project: &str) -> NewEvent {
    channel_event(project, "project", Some(project), Some(&format!("The {project} project")), &[])
}

fn channel_event(name: &str, kind: &str, project: Option<&str>, purpose: Option<&str>, members: &[String]) -> NewEvent {
    let id = crate::ids::ulid();
    NewEvent::system(
        "channel.created",
        json!({ "id": id, "name": name, "kind": kind, "project": project, "purpose": purpose, "members": members }),
    )
    .subject(id)
}

/// The name of the direct channel between two actors.
pub fn direct_name(a: &str, b: &str) -> String {
    let short = |x: &str| x.strip_prefix("agent:").unwrap_or(x).to_string();
    let mut v = [short(a), short(b)];
    v.sort();
    format!("dm:{}+{}", v[0], v[1])
}

/// Find a channel by `#name`, `name`, `@a+b` or id.
pub async fn channel(ledger: &Ledger, name: &str) -> Result<Option<Channel>> {
    let n = name.strip_prefix('#').unwrap_or(name);
    let n = match n.strip_prefix('@') {
        Some(pair) => format!("dm:{pair}"),
        None => n.to_string(),
    };
    Ok(sqlx::query_as("SELECT id, name, kind, project, purpose, members FROM channels WHERE name = ? OR id = ?")
        .bind(&n)
        .bind(&n)
        .fetch_optional(ledger.pool())
        .await?)
}

pub async fn channels(ledger: &Ledger) -> Result<Vec<Channel>> {
    Ok(sqlx::query_as("SELECT id, name, kind, project, purpose, members FROM channels ORDER BY kind = 'direct', name")
        .fetch_all(ledger.pool())
        .await?)
}

/// Whether `who` may read `ch`: direct channels are for their members.
pub fn can_read(who: &Principal, ch: &Channel) -> bool {
    ch.kind != "direct" || matches!(who, Principal::Owner { .. }) || ch.members().contains(&who.actor())
}

/// Whether `who` may post in `ch` (spec §10.1, §8.4).
pub fn can_post(who: &Principal, ch: &Channel) -> Result<(), String> {
    match (who, ch.kind.as_str()) {
        (Principal::Instance { .. }, _) => Ok(()),
        (_, "system") => Err("#system is Spectra's: guard rails, schedules and hooks post there".into()),
        (Principal::Owner { .. }, _) => Ok(()),
        (_, "direct") if !ch.members().contains(&who.actor()) => Err(format!("{} is a direct channel you are not in", ch.display())),
        (Principal::Agent { project, scopes, .. }, "project") => {
            if ch.project.as_deref() == Some(project) || scopes.iter().any(|s| s == "projects:all") {
                Ok(())
            } else {
                Err(format!("{} belongs to another project; post in #{project}, #general, or ask for projects:all", ch.display()))
            }
        }
        _ => Ok(()),
    }
}

/// Post a message, checking everything first (spec §10.3).
pub async fn post(ledger: &Ledger, who: &Author<'_>, mut p: Post, now_ms: u64) -> Result<Posted> {
    if p.body.trim().is_empty() {
        bail!("a message needs a body");
    }
    if p.body.len() > MAX_BODY {
        bail!("a message's body is at most 64 KiB; this is {} KiB", p.body.len() / 1024);
    }
    if p.payload.as_ref().is_some_and(|v| v.to_string().len() > MAX_BODY) {
        bail!("a message's payload is at most 64 KiB");
    }
    let kind = p.kind.take().unwrap_or_else(|| "chat".into());
    if !KINDS.contains(&kind.as_str()) {
        bail!("a message's kind is one of {}", KINDS.join(", "));
    }
    let ch = match (&p.channel, &p.reply_to, &who.principal) {
        (Some(c), _, _) => channel(ledger, c).await?.with_context(|| format!("no channel {c}"))?,
        (None, Some(r), _) => {
            let root = message(ledger, r).await?.with_context(|| format!("no message {r}"))?;
            channel(ledger, &root.channel).await?.context("the thread's channel")?
        }
        (None, None, Principal::Agent { project, .. }) => {
            channel(ledger, project).await?.with_context(|| format!("no channel for project {project}"))?
        }
        (None, None, _) => channel(ledger, "general").await?.context("no #general (spectra init makes it)")?,
    };
    can_post(&who.principal, &ch).map_err(anyhow::Error::msg)?;
    let reply_to = match &p.reply_to {
        Some(r) => {
            let m = message(ledger, r).await?.with_context(|| format!("no message {r}"))?;
            if m.channel != ch.id {
                bail!("{r} is in another channel");
            }
            // A reply to a reply joins the root's thread.
            Some(m.reply_to.unwrap_or(m.id))
        }
        None => None,
    };
    let author = who.principal.actor();
    let dup: Option<String> = sqlx::query_scalar("SELECT id FROM messages WHERE channel = ? AND author = ? AND body = ? AND at_ms >= ? AND redacted IS NULL ORDER BY seq DESC LIMIT 1")
        .bind(&ch.id)
        .bind(&author)
        .bind(&p.body)
        .bind(now_ms.saturating_sub(DUPLICATE_MS) as i64)
        .fetch_optional(ledger.pool())
        .await?;
    if let Some(id) = dup {
        return Ok(Posted::Duplicate { id });
    }
    let mut mentions = resolve_mentions(ledger, &p.body, &ch).await?;
    for m in p.mentions.drain(..) {
        if let Some(a) = resolve_name(ledger, &m).await? {
            mentions.push(a);
        }
    }
    mentions.retain(|m| m != &author);
    mentions.sort();
    mentions.dedup();
    let mut refs = parse_refs(&p.body);
    refs.append(&mut p.refs);
    refs.sort();
    refs.dedup();
    let id = crate::ids::ulid();
    let payload = json!({
        "id": id, "channel": ch.id, "reply_to": reply_to, "kind": kind, "body": p.body,
        "payload": p.payload, "mentions": mentions, "refs": refs,
    });
    let mut e = who.event("message.posted", payload).subject(id.clone());
    e.project = ch.project.clone();
    ledger.append_one(e).await?;
    Ok(Posted::Message { id, channel: ch.display() })
}

/// Make the direct channel between `who` and `other` when there is none.
pub async fn ensure_direct(ledger: &Ledger, who: &Principal, other: &str) -> Result<Channel> {
    let other = resolve_name(ledger, other).await?.with_context(|| format!("nobody is called {other}"))?;
    let name = direct_name(&who.actor(), &other);
    if let Some(c) = channel(ledger, &name).await? {
        return Ok(c);
    }
    let mut members = vec![who.actor(), other];
    members.sort();
    ledger.append_one(channel_event(&name, "direct", None, None, &members)).await?;
    channel(ledger, &name).await?.context("the direct channel")
}

pub async fn message(ledger: &Ledger, id: &str) -> Result<Option<Message>> {
    Ok(sqlx::query_as(&format!("{SELECT_MESSAGE} WHERE id = ? OR (length(?) >= 6 AND id LIKE ? || '%') ORDER BY seq LIMIT 1"))
        .bind(id)
        .bind(id)
        .bind(id)
        .fetch_optional(ledger.pool())
        .await?)
}

const SELECT_MESSAGE: &str =
    "SELECT id, channel, reply_to, author, kind, body, payload, mentions, refs, at_ms, seq, replies, redacted FROM messages";

/// A channel's top-level messages (each with its reply count), or a whole
/// thread, oldest first; `since` is a message id or a time in ms.
pub async fn read(
    ledger: &Ledger,
    who: &Principal,
    channel_name: Option<&str>,
    thread: Option<&str>,
    since: Option<&str>,
    limit: u32,
) -> Result<(Channel, Vec<Message>)> {
    let limit = limit.clamp(1, 500) as i64;
    let since_seq = match since {
        None => 0,
        Some(s) => match s.parse::<i64>() {
            Ok(ms) if ms > 1_000_000_000_000 => {
                sqlx::query_scalar("SELECT COALESCE(MIN(seq), 1 << 62) - 1 FROM messages WHERE at_ms > ?")
                    .bind(ms)
                    .fetch_one(ledger.pool())
                    .await?
            }
            _ => message(ledger, s).await?.map(|m| m.seq).with_context(|| format!("no message {s}"))?,
        },
    };
    let (ch, rows) = match thread {
        Some(t) => {
            let root = message(ledger, t).await?.with_context(|| format!("no message {t}"))?;
            let root_id = root.reply_to.clone().unwrap_or(root.id.clone());
            let ch = channel(ledger, &root.channel).await?.context("the thread's channel")?;
            let rows: Vec<Message> =
                sqlx::query_as(&format!("{SELECT_MESSAGE} WHERE (id = ? OR reply_to = ?) AND seq > ? ORDER BY seq LIMIT ?"))
                    .bind(&root_id)
                    .bind(&root_id)
                    .bind(since_seq)
                    .bind(limit)
                    .fetch_all(ledger.pool())
                    .await?;
            (ch, rows)
        }
        None => {
            let name = match (channel_name, who) {
                (Some(c), _) => c.to_string(),
                (None, Principal::Agent { project, .. }) => project.clone(),
                (None, _) => "general".into(),
            };
            let ch = channel(ledger, &name).await?.with_context(|| format!("no channel {name}"))?;
            // The newest `limit`, returned oldest first.
            let mut rows: Vec<Message> =
                sqlx::query_as(&format!("{SELECT_MESSAGE} WHERE channel = ? AND reply_to IS NULL AND seq > ? ORDER BY seq DESC LIMIT ?"))
                    .bind(&ch.id)
                    .bind(since_seq)
                    .bind(limit)
                    .fetch_all(ledger.pool())
                    .await?;
            rows.reverse();
            (ch, rows)
        }
    };
    if !can_read(who, &ch) {
        bail!("{} is a direct channel you are not in", ch.display());
    }
    Ok((ch, rows.into_iter().map(shown).collect()))
}

/// A redacted message shows only that it was.
fn shown(mut m: Message) -> Message {
    if let Some(why) = &m.redacted {
        m.body = format!("[redacted: {why}]");
        m.payload = None;
    }
    m
}

/// One item in a principal's inbox.
#[derive(Clone, Debug, Serialize)]
pub struct Item {
    pub id: String,
    pub kind: String,
    pub at_ms: i64,
    pub from: String,
    pub channel: String,
    pub summary: String,
    pub action_hint: String,
}

/// The unread items of `actor`, oldest first; `deliver` records that they
/// were handed over (`inbox.delivered`).
pub async fn inbox(ledger: &Ledger, actor: &str, deliver: Option<&Author<'_>>) -> Result<Vec<Item>> {
    let rows: Vec<InboxRow> = sqlx::query_as(
        "SELECT i.item AS id, i.kind, m.at_ms, m.author AS from_, c.name AS channel, m.body, m.redacted FROM inbox i JOIN messages m ON m.id = i.item JOIN channels c ON c.id = m.channel \
         WHERE i.principal = ? AND i.read = 0 ORDER BY i.seq",
    )
    .bind(actor)
    .fetch_all(ledger.pool())
    .await?;
    let items: Vec<Item> = rows
        .into_iter()
        .map(|InboxRow { id, kind, at_ms, from_: from, channel: ch, body, redacted }| {
            let channel = match ch.strip_prefix("dm:") {
                Some(p) => format!("@{p}"),
                None => format!("#{ch}"),
            };
            let summary = if redacted.is_some() { "[redacted]".into() } else { summarise(&body, 500) };
            Item {
                action_hint: format!("read(thread: \"{id}\") for context, then say(reply_to: \"{id}\", …)"),
                id,
                kind,
                at_ms,
                from,
                channel,
                summary,
            }
        })
        .collect();
    if let (Some(who), Some(last)) = (deliver, items.last()) {
        let undelivered: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM inbox WHERE principal = ? AND delivered = 0 AND read = 0")
            .bind(actor)
            .fetch_one(ledger.pool())
            .await?;
        if undelivered > 0 {
            ledger.append_one(who.event("inbox.delivered", json!({ "principal": actor, "upto": last.id }))).await?;
        }
    }
    Ok(items)
}

/// The number of unread items, for `_meta`.
pub async fn unread(ledger: &Ledger, actor: &str) -> Result<i64> {
    Ok(sqlx::query_scalar("SELECT COUNT(*) FROM inbox WHERE principal = ? AND read = 0").bind(actor).fetch_one(ledger.pool()).await?)
}

/// Mark items read, with a reaction on each if given.
pub async fn ack(ledger: &Ledger, who: &Author<'_>, ids: &[String], reaction: Option<&str>) -> Result<usize> {
    if let Some(r) = reaction {
        if !REACTIONS.contains(&r) {
            bail!("a reaction is one of {}", REACTIONS.join(", "));
        }
    }
    let actor = who.principal.actor();
    let mut events = Vec::new();
    let mut found = Vec::new();
    for id in ids {
        let m = message(ledger, id).await?.with_context(|| format!("no message {id}"))?;
        found.push(m.id.clone());
        if let Some(r) = reaction {
            events.push(who.event("message.reaction", json!({ "id": m.id, "reaction": r })).subject(m.id.clone()));
        }
    }
    events.push(who.event("inbox.read", json!({ "principal": actor, "ids": found })));
    let n = found.len();
    ledger.append(events).await?;
    Ok(n)
}

/// Redact a message: its author's, or anyone's by the owner.
pub async fn redact(ledger: &Ledger, who: &Author<'_>, id: &str, reason: &str) -> Result<()> {
    let m = message(ledger, id).await?.with_context(|| format!("no message {id}"))?;
    if m.author != who.principal.actor() && !matches!(who.principal, Principal::Owner { .. }) {
        bail!("only its author or the owner may redact a message");
    }
    if reason.trim().is_empty() {
        bail!("say why it is redacted");
    }
    ledger.append_one(who.event("message.redacted", json!({ "id": m.id, "reason": reason })).subject(m.id)).await?;
    Ok(())
}

#[derive(sqlx::FromRow)]
struct InboxRow {
    id: String,
    kind: String,
    at_ms: i64,
    from_: String,
    channel: String,
    body: String,
    redacted: Option<String>,
}

#[derive(sqlx::FromRow)]
struct HitRow {
    id: String,
    name: String,
    author: String,
    at_ms: i64,
    snippet: String,
    reply_to: Option<String>,
    kind: String,
    members: String,
}

/// A search hit (spec §13.1).
#[derive(Clone, Debug, Serialize)]
pub struct Hit {
    pub id: String,
    pub channel: String,
    pub author: String,
    pub at_ms: i64,
    pub snippet: String,
    pub thread: Option<String>,
}

/// Full-text search over messages the caller may read, newest first.
pub async fn search(
    ledger: &Ledger,
    who: &Principal,
    query: &str,
    channel_name: Option<&str>,
    author: Option<&str>,
    since_ms: Option<i64>,
    limit: u32,
) -> Result<Vec<Hit>> {
    if query.trim().is_empty() {
        bail!("search needs a query");
    }
    let ch = match channel_name {
        Some(c) => Some(channel(ledger, c).await?.with_context(|| format!("no channel {c}"))?.id),
        None => None,
    };
    let author = match author {
        Some(a) => Some(resolve_name(ledger, a).await?.unwrap_or_else(|| a.to_string())),
        None => None,
    };
    let rows: Vec<HitRow> = sqlx::query_as(
        "SELECT m.id, c.name, m.author, m.at_ms, snippet(messages_fts, 3, '[', ']', '…', 16) AS snippet, m.reply_to, c.kind, c.members \
         FROM messages_fts f JOIN messages m ON m.id = f.id JOIN channels c ON c.id = m.channel \
         WHERE messages_fts MATCH ? AND m.redacted IS NULL AND (? IS NULL OR m.channel = ?) AND (? IS NULL OR m.author = ?) AND (? IS NULL OR m.at_ms >= ?) \
         ORDER BY m.seq DESC LIMIT ?",
    )
    .bind(query)
    .bind(&ch)
    .bind(&ch)
    .bind(&author)
    .bind(&author)
    .bind(since_ms)
    .bind(since_ms)
    .bind(limit.clamp(1, 100) as i64)
    .fetch_all(ledger.pool())
    .await
    .map_err(|e| match e {
        sqlx::Error::Database(d) if d.message().contains("fts5") || d.message().contains("syntax") => {
            anyhow::anyhow!("the query does not parse: words, \"phrases\", prefix*, AND, OR, NOT ({})", d.message())
        }
        e => e.into(),
    })?;
    Ok(rows
        .into_iter()
        .filter(|r| {
            r.kind != "direct"
                || matches!(who, Principal::Owner { .. })
                || serde_json::from_str::<Vec<String>>(&r.members).unwrap_or_default().contains(&who.actor())
        })
        .map(|HitRow { id, name, author, at_ms, snippet, reply_to, .. }| Hit {
            channel: match name.strip_prefix("dm:") {
                Some(p) => format!("@{p}"),
                None => format!("#{name}"),
            },
            thread: reply_to.or(Some(id.clone())),
            id,
            author,
            at_ms,
            snippet,
        })
        .collect())
}

/// `@name` in a body, resolved to actors; `@here` is the channel's members
/// that are online, resolved now so the inbox stays a function of events.
async fn resolve_mentions(ledger: &Ledger, body: &str, ch: &Channel) -> Result<Vec<String>> {
    let mut out = Vec::new();
    for word in body.split(|c: char| c.is_whitespace() || ",.;:!?()[]".contains(c)) {
        let Some(name) = word.strip_prefix('@') else { continue };
        if name == "here" {
            let online: Vec<String> =
                sqlx::query_scalar("SELECT 'agent:' || agent FROM presence WHERE state IN ('online', 'busy', 'idle')")
                    .fetch_all(ledger.pool())
                    .await?;
            let members = ch.members();
            out.extend(online.into_iter().filter(|a| members.is_empty() || members.contains(a)));
        } else if let Some(a) = resolve_name(ledger, name).await? {
            out.push(a);
        }
    }
    Ok(out)
}

/// `claude-oxygen`, `agent:claude-oxygen`, `human`, `owner`, or the owner's
/// first name, lower-cased, to an actor.
pub async fn resolve_name(ledger: &Ledger, name: &str) -> Result<Option<String>> {
    let n = name.trim_start_matches('@').to_lowercase();
    if n == "human" || n == "owner" {
        return Ok(Some("human".into()));
    }
    let n = n.strip_prefix("agent:").unwrap_or(&n).to_string();
    let agent: Option<String> = sqlx::query_scalar("SELECT name FROM agents WHERE name = ?").bind(&n).fetch_optional(ledger.pool()).await?;
    if let Some(a) = agent {
        return Ok(Some(format!("agent:{a}")));
    }
    let owner: Option<String> =
        sqlx::query_scalar("SELECT name FROM identities WHERE role = 'owner' AND current = 1").fetch_optional(ledger.pool()).await?;
    let first = owner.and_then(|o| o.split_whitespace().next().map(str::to_lowercase));
    Ok((first.as_deref() == Some(n.as_str())).then(|| "human".into()))
}

/// `task:ID`, `decision:ID`, `oxy:change:HASH` and URLs in a body.
pub fn parse_refs(body: &str) -> Vec<String> {
    body.split(|c: char| c.is_whitespace() || "()[]<>,;\"'".contains(c))
        .map(|w| w.trim_end_matches(['.', ':', '!', '?']))
        .filter(|w| {
            w.starts_with("task:")
                || w.starts_with("decision:")
                || w.starts_with("oxy:change:")
                || w.starts_with("https://")
                || w.starts_with("http://")
        })
        .filter(|w| w.len() > 6)
        .map(String::from)
        .collect()
}

fn summarise(body: &str, max: usize) -> String {
    let one: String = body.split_whitespace().collect::<Vec<_>>().join(" ");
    if one.chars().count() <= max {
        one
    } else {
        format!("{}…", one.chars().take(max - 1).collect::<String>())
    }
}

/// Apply conversation events to their tables.
pub async fn project(conn: &mut SqliteConnection, e: &Event) -> sqlx::Result<()> {
    let p = e.json();
    let s = |k: &str| p.get(k).and_then(Json::as_str).map(String::from);
    let list = |k: &str| p.get(k).cloned().filter(Json::is_array).unwrap_or(json!([]));
    match e.kind.as_str() {
        "channel.created" if s("id").is_some() && s("name").is_some() => {
            sqlx::query("INSERT OR IGNORE INTO channels (id, name, kind, project, purpose, members, seq) VALUES (?, ?, ?, ?, ?, ?, ?)")
                .bind(s("id"))
                .bind(s("name"))
                .bind(s("kind").unwrap_or_else(|| "shared".into()))
                .bind(s("project"))
                .bind(s("purpose"))
                .bind(list("members").to_string())
                .bind(e.seq as i64)
                .execute(&mut *conn)
                .await?;
        }
        "message.posted" if s("id").is_some() && s("channel").is_some() && s("body").is_some() => {
            let id = s("id").unwrap_or_default();
            let ch = s("channel").unwrap_or_default();
            let body = s("body").unwrap_or_default();
            let inserted = sqlx::query(
                "INSERT OR IGNORE INTO messages (id, channel, reply_to, author, kind, body, payload, mentions, refs, at_ms, seq) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            )
            .bind(&id)
            .bind(&ch)
            .bind(s("reply_to"))
            .bind(&e.actor)
            .bind(s("kind").unwrap_or_else(|| "chat".into()))
            .bind(&body)
            .bind(p.get("payload").filter(|v| !v.is_null()).map(|v| v.to_string()))
            .bind(list("mentions").to_string())
            .bind(list("refs").to_string())
            .bind(e.at_ms as i64)
            .bind(e.seq as i64)
            .execute(&mut *conn)
            .await?
            .rows_affected();
            if inserted == 0 {
                return Ok(());
            }
            if let Some(root) = s("reply_to") {
                sqlx::query("UPDATE messages SET replies = replies + 1 WHERE id = ?").bind(root).execute(&mut *conn).await?;
            }
            sqlx::query("INSERT INTO messages_fts (id, channel, author, body) VALUES (?, ?, ?, ?)")
                .bind(&id)
                .bind(&ch)
                .bind(&e.actor)
                .bind(&body)
                .execute(&mut *conn)
                .await?;
            let mut targets: Vec<(String, &str)> =
                list("mentions").as_array().into_iter().flatten().filter_map(|m| m.as_str().map(|a| (a.to_string(), "mention"))).collect();
            let members: Option<(String, String)> =
                sqlx::query_as("SELECT kind, members FROM channels WHERE id = ?").bind(&ch).fetch_optional(&mut *conn).await?;
            if let Some((kind, members)) = members {
                if kind == "direct" {
                    for m in serde_json::from_str::<Vec<String>>(&members).unwrap_or_default() {
                        targets.push((m, "dm"));
                    }
                }
            }
            for (who, kind) in targets {
                if who == e.actor {
                    continue;
                }
                sqlx::query("INSERT OR IGNORE INTO inbox (principal, item, kind, seq) VALUES (?, ?, ?, ?)")
                    .bind(who)
                    .bind(&id)
                    .bind(kind)
                    .bind(e.seq as i64)
                    .execute(&mut *conn)
                    .await?;
            }
        }
        "message.redacted" if s("id").is_some() => {
            sqlx::query("UPDATE messages SET redacted = ? WHERE id = ?")
                .bind(s("reason").unwrap_or_else(|| "redacted".into()))
                .bind(s("id"))
                .execute(&mut *conn)
                .await?;
            sqlx::query("DELETE FROM messages_fts WHERE id = ?").bind(s("id")).execute(&mut *conn).await?;
        }
        "message.reaction" if s("id").is_some() && s("reaction").is_some() => {
            sqlx::query("INSERT OR IGNORE INTO reactions (message, actor, reaction) VALUES (?, ?, ?)")
                .bind(s("id"))
                .bind(&e.actor)
                .bind(s("reaction"))
                .execute(&mut *conn)
                .await?;
        }
        "inbox.delivered" if s("principal").is_some() => {
            let upto: Option<i64> = sqlx::query_scalar("SELECT seq FROM inbox WHERE principal = ? AND item = ?")
                .bind(s("principal"))
                .bind(s("upto"))
                .fetch_optional(&mut *conn)
                .await?;
            if let Some(seq) = upto {
                sqlx::query("UPDATE inbox SET delivered = 1 WHERE principal = ? AND seq <= ?")
                    .bind(s("principal"))
                    .bind(seq)
                    .execute(&mut *conn)
                    .await?;
            }
        }
        "inbox.read" if s("principal").is_some() => {
            for id in list("ids").as_array().into_iter().flatten().filter_map(Json::as_str) {
                sqlx::query("UPDATE inbox SET read = 1, delivered = 1 WHERE principal = ? AND item = ?")
                    .bind(s("principal"))
                    .bind(id)
                    .execute(&mut *conn)
                    .await?;
            }
        }
        _ => {}
    }
    Ok(())
}

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

    #[test]
    fn refs_and_direct_names() {
        assert_eq!(
            parse_refs("see task:01ABC and oxy:change:abcdef12, https://oxygenhub.io/x."),
            ["task:01ABC", "oxy:change:abcdef12", "https://oxygenhub.io/x"]
        );
        assert_eq!(direct_name("agent:zed", "human"), "dm:human+zed");
        assert_eq!(direct_name("human", "agent:zed"), direct_name("agent:zed", "human"));
        assert_eq!(summarise("a\n  b", 10), "a b");
        assert_eq!(summarise("abcdefghijkl", 5), "abcd…");
    }
}
`/`	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-core/src/conversation.rs 89.5%

Keyboard
