Channels, threads, messages, the inbox and full-text search, shared by every interface (#6) 5eb57578f4ef

-- Conversation (spec §10): channels, messages and threads, reactions, each

-- principal's inbox, and full-text search over messages (§13.1).

CREATE TABLE channels (

    id        TEXT PRIMARY KEY,

    name      TEXT NOT NULL UNIQUE,     -- general, humans, system, a project's name, or dm:A+B

    kind      TEXT NOT NULL,            -- project | shared | direct | system

    project   TEXT,

    purpose   TEXT,

    members   TEXT NOT NULL,            -- JSON array of actors; empty: everyone

    seq       INTEGER NOT NULL

);

CREATE TABLE messages (

    id              TEXT PRIMARY KEY,

    channel         TEXT NOT NULL,

    reply_to        TEXT,

    author          TEXT NOT NULL,      -- the event's actor

    kind            TEXT NOT NULL,

    body            TEXT NOT NULL,

    payload         TEXT,

    mentions        TEXT NOT NULL,      -- JSON array of actors

    refs            TEXT NOT NULL,      -- JSON array

    at_ms           INTEGER NOT NULL,

    seq             INTEGER NOT NULL,

    replies         INTEGER NOT NULL DEFAULT 0,

    redacted        TEXT                -- the reason, once redacted

);

CREATE INDEX messages_channel ON messages(channel, seq);

CREATE INDEX messages_thread ON messages(reply_to, seq);

CREATE TABLE reactions (

    message   TEXT NOT NULL,

    actor     TEXT NOT NULL,

    reaction  TEXT NOT NULL,

    PRIMARY KEY (message, actor, reaction)

);

-- What each principal has to look at: mentions and direct messages for now.

CREATE TABLE inbox (

    principal  TEXT NOT NULL,

    item       TEXT NOT NULL,           -- the message id

    kind       TEXT NOT NULL,           -- mention | dm

    seq        INTEGER NOT NULL,

    delivered  INTEGER NOT NULL DEFAULT 0,

    read       INTEGER NOT NULL DEFAULT 0,

    PRIMARY KEY (principal, item)

);

CREATE VIRTUAL TABLE messages_fts USING fts5(id UNINDEXED, channel UNINDEXED, author UNINDEXED, body, tokenize = 'porter unicode61');

//! 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…");

    }

}

//! A world for scenario tests: an owner, the instance, the default channels,

//! two projects with their channels, and agents the owner delegated to, all

//! recorded as `spectra init`, `project add` and `agent add` record them.

#![allow(dead_code)]

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

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

use spectra_core::conversation::{default_channel_events, project_channel_event, Author, Signer};

use spectra_core::ids::now_ms;

use spectra_core::keys::KeyPair;

use spectra_core::ledger::Ledger;

use spectra_core::principals::{identity_payload, signed_event, Principal, Principals, DELEGATION_SCOPE};

use std::path::PathBuf;

pub struct World {

    pub dir: tempfile::TempDir,

    pub owner_home: PathBuf,

    pub ledger: Ledger,

    pub owner: KeyPair,

}

impl World {

    pub async fn new(projects: &[&str]) -> World {

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

        let owner_home = dir.path().join("owner");

        let owner = KeyPair::create_home(&owner_home, "Andy McCall", "[email protected]").unwrap();

        let data = dir.path().join("data");

        std::fs::create_dir_all(&data).unwrap();

        let instance = KeyPair::instance(&data, "laptop").unwrap();

        let ledger = Ledger::open(&data.join("spectra.db"), Some(instance.clone())).await.unwrap();

        let mut first = vec![

            signed_event(&owner, "human", "identity.owner", identity_payload(&owner.identity)),

            signed_event(&instance, "system", "identity.instance", identity_payload(&instance.identity)),

        ];

        first.extend(default_channel_events());

        ledger.append(first).await.unwrap();

        for p in projects {

            let payload = serde_json::json!({ "name": p, "path": format!("/src/{p}"), "modules": [] });

            ledger.append(vec![signed_event(&owner, "human", "project.added", payload), project_channel_event(p)]).await.unwrap();

        }

        World { dir, owner_home, ledger, owner }

    }

    /// Delegate to a new agent and add it, as `spectra agent add` does.

    pub async fn agent(&mut self, name: &str, project: &str) -> KeyPair {

        let agent = KeyPair::generate(name, "");

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

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

        let mut d = self.owner.identity.delegates.clone();

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

        let next = self.owner.identity.successor(d, None, now_ms());

        let k = next.hash();

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

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

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

        self.owner = KeyPair::load_home(&self.owner_home).unwrap();

        let mut rotated = identity_payload(&self.owner.identity);

        rotated["chain"] = serde_json::json!([]);

        let mut p = identity_payload(&agent.identity);

        p["name"] = name.into();

        p["kind"] = "claude-code".into();

        p["project"] = project.into();

        p["owner_name"] = "Andy McCall".into();

        p["scopes"] = spectra_core::principals::DEFAULT_SCOPES.into();

        self.ledger

            .append(vec![

                signed_event(&self.owner, "human", "identity.rotated", rotated),

                signed_event(&self.owner, "human", "agent.added", p),

            ])

            .await

            .unwrap();

        agent

    }

    pub async fn principals(&self) -> Principals {

        Principals::load(&self.ledger).await.unwrap()

    }

    /// An author signing from the command line, as the CLI does.

    pub async fn author<'a>(&self, pair: &'a KeyPair) -> Author<'a> {

        let principal: Principal = self.principals().await.principal(&pair.key, now_ms()).unwrap();

        Author { principal, signer: Signer::Cli(pair) }

    }

    pub async fn verify_clean(&self) {

        let v = self.ledger.verify(&mut Principals::default()).await.unwrap();

        assert!(v.clean(), "{:?}", v.findings);

        assert!(self.ledger.check().await.unwrap().is_empty(), "projections drifted");

    }

}

//! Conversation (spec §10, §13.1): who may post where, threads, mentions

//! and direct messages into the inbox, duplicates, limits, redaction, and

//! search limited to what the reader may see.

mod common;

use common::World;

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

use spectra_core::ids::now_ms;

fn post(body: &str) -> Post {

    Post { body: body.into(), ..Default::default() }

}

fn id(p: Posted) -> String {

    match p {

        Posted::Message { id, .. } => id,

        Posted::Duplicate { id } => panic!("dropped as a duplicate of {id}"),

    }

}

#[tokio::test]

async fn agents_post_where_they_belong_and_nowhere_else() {

    let mut w = World::new(&["oxygen", "oxygenhub"]).await;

    let claude = w.agent("claude-oxygen", "oxygen").await;

    let a = w.author(&claude).await;

    let at = conv::post(&w.ledger, &a, post("default goes to my project"), now_ms()).await.unwrap();

    assert!(matches!(&at, Posted::Message { channel, .. } if channel == "#oxygen"), "{at:?}");

    for c in ["general", "humans", "#oxygen"] {

        conv::post(&w.ledger, &a, Post { channel: Some(c.into()), ..post(&format!("hello {c}")) }, now_ms()).await.unwrap();

    }

    for (c, why) in [("oxygenhub", "another project"), ("system", "Spectra's")] {

        let err = conv::post(&w.ledger, &a, Post { channel: Some(c.into()), ..post("nope") }, now_ms()).await.unwrap_err();

        assert!(err.to_string().contains(why), "{c}: {err}");

    }

    let owner_pair = w.owner.clone();

    let owner = w.author(&owner_pair).await;

    conv::post(&w.ledger, &owner, Post { channel: Some("oxygenhub".into()), ..post("the owner posts anywhere but #system") }, now_ms())

        .await

        .unwrap();

    w.verify_clean().await;

}

#[tokio::test]

async fn threads_mentions_and_direct_messages_reach_the_inbox() {

    let mut w = World::new(&["oxygen", "oxygenhub"]).await;

    let claude = w.agent("claude-oxygen", "oxygen").await;

    let kimi = w.agent("kimi-hub", "oxygenhub").await;

    let (c, k) = (w.author(&claude).await, w.author(&kimi).await);

    let root = id(conv::post(

        &w.ledger,

        &c,

        Post { channel: Some("general".into()), ..post("@kimi-hub the wire protocol grew a field; see oxy:change:abcdef12") },

        now_ms(),

    )

    .await

    .unwrap());

    let items = conv::inbox(&w.ledger, "agent:kimi-hub", Some(&k)).await.unwrap();

    assert_eq!(items.len(), 1);

    assert_eq!((items[0].kind.as_str(), items[0].from.as_str(), items[0].channel.as_str()), ("mention", "agent:claude-oxygen", "#general"));

    let reply = id(conv::post(&w.ledger, &k, Post { reply_to: Some(root.clone()), ..post("@andy should I bump the pin?") }, now_ms())

        .await

        .unwrap());

    let nested =

        id(conv::post(&w.ledger, &c, Post { reply_to: Some(reply.clone()), ..post("yes, after the release") }, now_ms()).await.unwrap());

    let (_, thread) = conv::read(&w.ledger, &c.principal, None, Some(&nested), None, 50).await.unwrap();

    assert_eq!(

        thread.iter().map(|m| m.id.clone()).collect::<Vec<_>>(),

        [root.clone(), reply.clone(), nested],

        "a reply to a reply joins the root's thread"

    );

    let (_, roots) = conv::read(&w.ledger, &c.principal, Some("general"), None, None, 50).await.unwrap();

    assert_eq!(roots.len(), 1);

    assert_eq!(roots[0].replies, 2);

    assert_eq!(conv::unread(&w.ledger, "human").await.unwrap(), 1, "@andy is the owner");

    assert_eq!(conv::ack(&w.ledger, &k, std::slice::from_ref(&root), Some("ack")).await.unwrap(), 1);

    assert!(conv::inbox(&w.ledger, "agent:kimi-hub", None).await.unwrap().is_empty());

    let dm = conv::ensure_direct(&w.ledger, &c.principal, "kimi-hub").await.unwrap();

    assert_eq!(dm.display(), "@claude-oxygen+kimi-hub");

    conv::post(&w.ledger, &c, Post { channel: Some(dm.name.clone()), ..post("just between us") }, now_ms()).await.unwrap();

    assert_eq!(conv::inbox(&w.ledger, "agent:kimi-hub", None).await.unwrap()[0].kind, "dm");

    let third = w.agent("third", "oxygen").await;

    let t = w.author(&third).await;

    assert!(conv::read(&w.ledger, &t.principal, Some(&dm.name), None, None, 10).await.is_err(), "not a member");

    assert!(conv::post(&w.ledger, &t, Post { channel: Some(dm.name.clone()), ..post("let me in") }, now_ms()).await.is_err());

    w.verify_clean().await;

}

#[tokio::test]

async fn duplicates_limits_and_redaction() {

    let mut w = World::new(&["oxygen"]).await;

    let claude = w.agent("claude-oxygen", "oxygen").await;

    let c = w.author(&claude).await;

    let first = id(conv::post(&w.ledger, &c, post("build failed on main"), now_ms()).await.unwrap());

    assert!(

        matches!(conv::post(&w.ledger, &c, post("build failed on main"), now_ms()).await.unwrap(), Posted::Duplicate { id } if id == first)

    );

    assert!(conv::post(&w.ledger, &c, post(&"x".repeat(conv::MAX_BODY + 1)), now_ms()).await.is_err());

    assert!(conv::post(&w.ledger, &c, post("   "), now_ms()).await.is_err());

    assert!(conv::post(&w.ledger, &c, Post { kind: Some("shout".into()), ..post("x") }, now_ms()).await.is_err());

    let secret = id(conv::post(&w.ledger, &c, post("the token is hunter2"), now_ms()).await.unwrap());

    assert_eq!(conv::search(&w.ledger, &c.principal, "hunter2", None, None, None, 10).await.unwrap().len(), 1);

    let kimi = w.agent("kimi", "oxygen").await;

    let k = w.author(&kimi).await;

    assert!(conv::redact(&w.ledger, &k, &secret, "not mine").await.is_err(), "only its author or the owner");

    conv::redact(&w.ledger, &c, &secret, "a secret pasted by mistake").await.unwrap();

    let (_, msgs) = conv::read(&w.ledger, &c.principal, None, None, None, 10).await.unwrap();

    let shown = msgs.iter().find(|m| m.id == secret).unwrap();

    assert_eq!(shown.body, "[redacted: a secret pasted by mistake]");

    assert!(conv::search(&w.ledger, &c.principal, "hunter2", None, None, None, 10).await.unwrap().is_empty(), "gone from search");

    w.verify_clean().await;

}

#[tokio::test]

async fn search_finds_stemmed_words_and_respects_who_may_read() {

    let mut w = World::new(&["oxygen"]).await;

    let claude = w.agent("claude-oxygen", "oxygen").await;

    let kimi = w.agent("kimi", "oxygen").await;

    let (c, k) = (w.author(&claude).await, w.author(&kimi).await);

    conv::post(

        &w.ledger,

        &c,

        Post { channel: Some("general".into()), ..post("We decided to bump the oxygen pin after releasing 0.11") },

        now_ms(),

    )

    .await

    .unwrap();

    conv::post(&w.ledger, &c, post("Unrelated chatter about lunch"), now_ms()).await.unwrap();

    let dm = conv::ensure_direct(&w.ledger, &c.principal, "human").await.unwrap();

    conv::post(&w.ledger, &c, Post { channel: Some(dm.name.clone()), ..post("privately: the pin bump worries me") }, now_ms())

        .await

        .unwrap();

    let hits = conv::search(&w.ledger, &k.principal, "bumping pins", None, None, None, 20).await.unwrap();

    assert_eq!(hits.len(), 1, "stemmed, and the direct message is not kimi's to see: {hits:?}");

    assert!(hits[0].snippet.contains("[bump]"), "{}", hits[0].snippet);

    assert_eq!(hits[0].channel, "#general");

    let owner_pair = w.owner.clone();

    let owner = w.author(&owner_pair).await;

    assert_eq!(

        conv::search(&w.ledger, &owner.principal, "pin", None, None, None, 20).await.unwrap().len(),

        2,

        "the owner reads everything"

    );

    assert_eq!(

        conv::search(&w.ledger, &k.principal, "\"oxygen pin\"", Some("general"), Some("claude-oxygen"), None, 20).await.unwrap().len(),

        1

    );

    assert!(conv::search(&w.ledger, &k.principal, "lunch", Some("general"), None, None, 20).await.unwrap().is_empty());

    let err = conv::search(&w.ledger, &k.principal, "AND OR (", None, None, None, 20).await.unwrap_err();

    assert!(err.to_string().contains("does not parse"), "{err}");

    w.verify_clean().await;

}

- **Added: channels, threads, messages, the inbox and search (#6).** `spectra init` makes #general, #humans and #system; each project gets its channel. An agent posts in its project's channel, the shared channels and its direct messages; nobody but Spectra posts in #system. Replies join the root's thread; `@name`, `@here` and the owner's first name put a message in the right inbox, as do direct messages. Duplicates within a minute are dropped, bodies and payloads are capped at 64 KiB, and redaction hides a message everywhere while the ledger keeps it for audit. Full-text search (FTS5, stemmed) covers what the reader may see.

pub mod conversation;

pub const TABLES: &[&str] = &[

    "event_counts",

    "identities",

    "agents",

    "projects",

    "project_relations",

    "presence",

    "sessions",

    "channels",

    "messages",

    "reactions",

    "inbox",

    "messages_fts",

];

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

- [x] #6 Channels, threads, messages, inbox and search — done 2026-11-02: spectra-core::conversation (post, read, inbox, ack, redact, search, direct channels), shared by tools, API and CLI.

    let mut first = vec![

        signed_event(&owner, "human", "identity.owner", identity_payload(&owner.identity)),

        signed_event(&instance, "system", "identity.instance", identity_payload(&instance.identity)),

    ];

    first.extend(spectra_core::conversation::default_channel_events());

    ledger.append(first).await?;

    ledger

        .append(vec![

            signed_event(&owner, "human", "project.added", payload).project(name.clone()).subject(name.clone()),

            spectra_core::conversation::project_channel_event(&name),

        ])

        .await?;