team_core/

mailbox.rs

//! SQLite mailbox schema shared by `team-mcp` and integration tests.
//!
//! The actual connection handling lives in `team-mcp`; this module defines
//! the schema + migrations so both crates agree on the shape of the data.

/// The one privileged mailbox `kind`. A `system` message is a lifecycle signal
/// (drain, startup, rate-limit) the supervisor emits inline + real-time; only a
/// `system:*` source may originate one (#254). If any agent or `user:*` could,
/// a forged "session terminating" signal would be trivial.
pub const PRIVILEGED_KIND: &str = "system";

/// Is `kind` the privileged system kind? Single source of truth for the
/// privileged-kind contract, consulted on every mailbox *write* path — the
/// insert allowlist (`team-mcp` `store::send_dm_kind`, sender-gated) and the
/// UPDATE guard (`team-bot` media dispatch, which refuses it outright) — so the
/// contract has one definition rather than a `"system"` literal copied per site
/// (#320).
pub fn is_privileged_kind(kind: &str) -> bool {
    kind == PRIVILEGED_KIND
}

/// Idempotent schema bootstrap. Safe to run on every connect.
pub const SCHEMA: &str = r#"
-- NOTE: pragmas (journal_mode=WAL, busy_timeout, foreign_keys) are set by
-- the connection opener *before* this batch runs — concurrent openers race
-- if we set them here.

CREATE TABLE IF NOT EXISTS projects (
    id   TEXT PRIMARY KEY,
    name TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS agents (
    id         TEXT PRIMARY KEY,          -- "<project>:<agent>"
    project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
    role       TEXT NOT NULL,
    runtime    TEXT NOT NULL,
    is_manager INTEGER NOT NULL DEFAULT 0,
    reports_to TEXT                        -- short name, resolved within project
);

CREATE INDEX IF NOT EXISTS agents_project_idx ON agents(project_id);

CREATE TABLE IF NOT EXISTS messages (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id   TEXT NOT NULL,
    sender       TEXT NOT NULL,            -- "<project>:<agent>" or "user:<handle>" or "cli"
    recipient    TEXT NOT NULL,            -- "<project>:<agent>" or "channel:<project>:<name>"
    text         TEXT NOT NULL,
    thread_id    TEXT,
    sent_at      REAL NOT NULL,
    delivered_at REAL,
    acked_at     REAL
);

CREATE INDEX IF NOT EXISTS messages_recipient_idx
    ON messages(recipient, acked_at);
CREATE INDEX IF NOT EXISTS messages_project_idx
    ON messages(project_id, sent_at);

-- Channels + subscriptions + per-agent ACLs.
CREATE TABLE IF NOT EXISTS channels (
    id         TEXT PRIMARY KEY,               -- "<project>:<name>"
    project_id TEXT NOT NULL,
    name       TEXT NOT NULL,
    wildcard   INTEGER NOT NULL DEFAULT 0       -- 1 iff members = "*"
);

CREATE TABLE IF NOT EXISTS channel_members (
    channel_id TEXT NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
    agent_id   TEXT NOT NULL,
    PRIMARY KEY (channel_id, agent_id)
);

CREATE INDEX IF NOT EXISTS channel_members_agent_idx
    ON channel_members(agent_id);

CREATE TABLE IF NOT EXISTS agent_acls (
    agent_id        TEXT PRIMARY KEY REFERENCES agents(id) ON DELETE CASCADE,
    can_dm_json     TEXT NOT NULL DEFAULT '[]',    -- ["dev","critic"]
    can_bcast_json  TEXT NOT NULL DEFAULT '[]'     -- ["product","all"]
);

-- Inter-project manager bridges.
CREATE TABLE IF NOT EXISTS bridges (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    from_agent   TEXT NOT NULL,             -- "<project>:<agent>", must be a manager
    to_agent     TEXT NOT NULL,             -- "<project>:<agent>", must be a manager
    topic        TEXT NOT NULL,
    opened_by    TEXT NOT NULL,             -- "user:<handle>" or "cli"
    opened_at    REAL NOT NULL,
    expires_at   REAL NOT NULL,
    closed_at    REAL
);

CREATE INDEX IF NOT EXISTS bridges_open_idx
    ON bridges(expires_at, closed_at);

-- Human-in-the-loop permission fabric.
CREATE TABLE IF NOT EXISTS approvals (
    id             INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id     TEXT NOT NULL,
    agent_id       TEXT NOT NULL,
    action         TEXT NOT NULL,          -- "publish", "deploy", ...
    scope_tag      TEXT,                   -- optional narrower tag
    summary        TEXT NOT NULL,
    payload_json   TEXT,
    status         TEXT NOT NULL,          -- pending | approved | denied | expired | undeliverable
    requested_at   REAL NOT NULL,
    decided_at     REAL,
    decided_by     TEXT,
    decision_note  TEXT,
    expires_at     REAL NOT NULL,
    delivered_at   REAL                    -- NULL until an interface adapter confirms surfacing to a human
);

CREATE INDEX IF NOT EXISTS approvals_pending_idx
    ON approvals(status, expires_at);

-- Budget ledger. Rows are appended by runtime cost parsers (the claude-code
-- writer is `teamctl budget-record`, fired from the Stop hook). `teamctl
-- budget` aggregates per project/day.
CREATE TABLE IF NOT EXISTS budget (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id  TEXT NOT NULL,
    agent_id    TEXT,
    runtime     TEXT,
    usd         REAL NOT NULL DEFAULT 0,
    -- claude-code writer: prompt + both cache buckets (the schema has no cache
    -- columns; cache tokens are priced into `usd` at their own rates). So this
    -- is "all non-output tokens", not strictly Anthropic "input tokens".
    input_tok   INTEGER NOT NULL DEFAULT 0,
    output_tok  INTEGER NOT NULL DEFAULT 0,
    observed_at REAL NOT NULL
);

CREATE INDEX IF NOT EXISTS budget_project_day_idx
    ON budget(project_id, observed_at);

-- Rate-limit events. Written by `teamctl rl-watch` whenever a runtime
-- emits a rate-limit signature. Hooks (notify, webhook, run) run off these
-- rows; the wrapper loop sleeps until `resets_at` before respawning.
CREATE TABLE IF NOT EXISTS rate_limits (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    agent_id    TEXT NOT NULL,
    runtime     TEXT NOT NULL,
    hit_at      REAL NOT NULL,
    resets_at   REAL,                  -- nullable: sometimes we can't parse
    raw_match   TEXT NOT NULL,
    handled_at  REAL
);

CREATE INDEX IF NOT EXISTS rate_limits_agent_idx
    ON rate_limits(agent_id, hit_at);
"#;

/// Bootstrap the schema and apply additive migrations. Idempotent — safe on
/// every connect. Replaces direct `execute_batch(SCHEMA)` calls so that
/// existing databases pick up new columns without a destructive reset.
pub fn ensure(conn: &rusqlite::Connection) -> rusqlite::Result<()> {
    conn.execute_batch(SCHEMA)?;
    // Additive migrations. SQLite has no `ADD COLUMN IF NOT EXISTS`, so each
    // migration tolerates the "duplicate column name" error to stay idempotent.
    let migrations: &[&str] = &[
        "ALTER TABLE approvals ADD COLUMN delivered_at REAL",
        // T-086-A: discriminator + structured payload for non-text mailbox kinds
        // (image, file, reaction). Existing text rows have NULL on both — readers
        // treat NULL kind as 'text' for back-compat.
        // #254/#320: `'system'` is also a recognized kind value — lifecycle/
        // system signals (drain, startup, rate-limit) the supervisor emits. No
        // schema change is needed: `kind` is free-form TEXT with no CHECK/enum
        // and the db has no version mechanism, so the value's contract lives in
        // code (see `is_privileged_kind`): originated only by a `system:*`
        // source at the insert choke point (store::send_dm_kind), refused on
        // every UPDATE path (team-bot media dispatch), and always-inline channel
        // delivery (team-mcp format_channel_event), never a lazy stub.
        "ALTER TABLE messages ADD COLUMN kind TEXT",
        "ALTER TABLE messages ADD COLUMN structured_payload TEXT",
        // T-086-B: Telegram message id this row pertains to. Direction-
        // disambiguated by sender: inbound rows (sender = `user:telegram`)
        // store the source Telegram message id so agents know what to
        // reply to; outbound rows (sender = `<project>:<agent>`) store the
        // id this reply threads under for `reply_parameters`. NULL on
        // pre-T-086-B rows and on rows that aren't Telegram-bound.
        "ALTER TABLE messages ADD COLUMN telegram_msg_id INTEGER",
        // T-104: per-message delivery mode for lazy inbox. NULL = lazy
        // (the channel watcher emits a stub; the agent drills in via
        // `inbox_read`). `'immediate'` = full body delivered inline,
        // bypassing the stub. Set by the bot when an operator prefixes a
        // message with `/readnow `.
        "ALTER TABLE messages ADD COLUMN delivery_mode TEXT",
        // #299: multi-option interactive decisions. `options_json` is a
        // JSON array of `{label,value}` the bot renders as N inline
        // buttons; NULL means the binary Approve/Deny back-compat path
        // (existing callers never set options). `decision_value` holds
        // the chosen option's `value` once the operator taps; NULL for
        // binary decisions and for Cancel (status carries those).
        "ALTER TABLE approvals ADD COLUMN options_json TEXT",
        "ALTER TABLE approvals ADD COLUMN decision_value TEXT",
    ];
    for stmt in migrations {
        if let Err(e) = conn.execute(stmt, []) {
            let msg = e.to_string();
            if !msg.contains("duplicate column name") {
                return Err(e);
            }
        }
    }
    Ok(())
}

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

    #[test]
    fn privileged_kind_const_is_system() {
        // The privileged-kind contract is pinned to the literal `'system'`
        // (#254/#320). A rename here would silently desync the insert
        // allowlist and the UPDATE guard, so pin the value itself.
        assert_eq!(PRIVILEGED_KIND, "system");
    }

    #[test]
    fn is_privileged_kind_true_only_for_system() {
        assert!(is_privileged_kind("system"));
    }

    #[test]
    fn is_privileged_kind_false_for_ordinary_kinds() {
        // Every kind a non-`system:*` source legitimately writes — the
        // media UPDATE kinds, plain text, and the empty/NULL-as-text
        // sentinel — must be allowed through, or the guard would refuse
        // ordinary traffic (#320).
        for kind in ["image", "file", "media_error", "text", ""] {
            assert!(
                !is_privileged_kind(kind),
                "kind {kind:?} must not be privileged"
            );
        }
    }
}