The human at a terminal: spectra say, inbox --follow, search and api, signed with the owner's key (#12) da84e98f104e

//! The human at a terminal (spec §19): `spectra say`, `spectra inbox

//! [--follow]`, `spectra search`, and `spectra api` for anything else. Each

//! signs with the owner's own key: through the daemon when it is serving,

//! else straight into the ledger.

use super::identity::{owner_home, recorded_owner};

use super::{ui, Ctx};

use anyhow::{bail, Context, Result};

use serde_json::{json, Value};

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

use spectra_core::keys::KeyPair;

use spectra_core::ledger::Ledger;

use spectra_core::principals::Principals;

use spectra_mcp::client::Client;

use std::path::PathBuf;

/// A client signing as the owner, when the daemon is serving.

async fn owner_client(ctx: &Ctx, home: &std::path::Path) -> Result<Option<Client>> {

    if !Client::alive(&ctx.socket()).await {

        return Ok(None);

    }

    let ledger = Ledger::open_read_only(&ctx.db()?).await?;

    Ok(Some(Client::new(&ctx.socket(), Some(recorded_owner(&ledger, home).await?))))

}

async fn api(c: &Client, method: &str, path: &str, body: Option<Value>) -> Result<Value> {

    let r = c.send(method, path, &[], body.map(|b| serde_json::to_vec(&b)).transpose()?.unwrap_or_default()).await?;

    let v: Value = r.json().unwrap_or(Value::Null);

    if r.status >= 400 {

        let why = v

            .get("detail")

            .or(v.get("title"))

            .and_then(Value::as_str)

            .map(String::from)

            .unwrap_or_else(|| String::from_utf8_lossy(&r.body).into());

        bail!("{why}");

    }

    Ok(v)

}

fn encode(s: &str) -> String {

    s.bytes()

        .map(|b| match b {

            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => (b as char).to_string(),

            _ => format!("%{b:02X}"),

        })

        .collect()

}

pub async fn say(

    ctx: &Ctx,

    channel: Option<String>,

    reply_to: Option<String>,

    body: String,

    owner_home_flag: Option<PathBuf>,

    as_json: bool,

) -> Result<i32> {

    let home = owner_home(ctx, owner_home_flag);

    let (id, where_) = match owner_client(ctx, &home).await? {

        Some(c) => {

            let ch = channel.clone().unwrap_or_else(|| "general".into());

            let v = api(

                &c,

                "POST",

                &format!("/api/v1/channels/{}/messages", encode(ch.trim_start_matches('#'))),

                Some(json!({ "body": body, "reply_to": reply_to })),

            )

            .await?;

            if v["duplicate"] == true {

                bail!("you said exactly this in the last minute ({})", v["id"].as_str().unwrap_or(""));

            }

            (v["id"].as_str().unwrap_or("").to_string(), v["channel"].as_str().unwrap_or("").to_string())

        }

        None => {

            let ledger = ctx.writer().await?;

            let owner = KeyPair::load_home(&home)?;

            let principal = Principals::load(&ledger)

                .await?

                .principal(&owner.key, spectra_core::ids::now_ms())

                .map_err(anyhow::Error::msg)

                .context("run `spectra identity refresh` first")?;

            let author = Author { principal, signer: Signer::Cli(&owner) };

            let post = Post {

                channel: if reply_to.is_some() { None } else { Some(channel.unwrap_or_else(|| "general".into())) },

                body,

                reply_to,

                ..Default::default()

            };

            let posted = conv::post(&ledger, &author, post, spectra_core::ids::now_ms()).await?;

            ledger.checkpoint().await?;

            match posted {

                Posted::Message { id, channel } => (id, channel),

                Posted::Duplicate { id } => bail!("you said exactly this in the last minute ({id})"),

            }

        }

    };

    if as_json {

        println!("{}", json!({ "id": id, "channel": where_ }));

    } else {

        println!("{} {}{}{}", ui::done("Posted"), ui::subject(&where_), ui::sep(), ui::id(&id));

    }

    Ok(0)

}

fn print_item(i: &Value) {

    let kind = i["kind"].as_str().unwrap_or("");

    let mark = if kind == "dm" { ui::warn(ui::glyph("✉", "@")) } else { ui::ok(ui::glyph("●", "*")) };

    let from = i["from_display"].as_str().or(i["from"].as_str()).unwrap_or("");

    let at = i["at_ms"].as_u64().map(ui::ago).unwrap_or_default();

    println!(

        "{mark} {} {}{}{}{}{}",

        ui::dim(kind),

        ui::author(from),

        ui::sep(),

        i["channel"].as_str().unwrap_or(""),

        ui::sep(),

        ui::dim(at)

    );

    println!("  {}", i["summary"].as_str().unwrap_or(""));

    println!("  {}", ui::dim(format!("reply: spectra say --reply-to {} \"…\"", i["id"].as_str().unwrap_or(""))));

}

pub async fn inbox(ctx: &Ctx, follow: bool, owner_home_flag: Option<PathBuf>, as_json: bool) -> Result<i32> {

    let home = owner_home(ctx, owner_home_flag);

    let ledger = Ledger::open_read_only(&ctx.db()?).await?;

    let mut items: Vec<Value> = Vec::new();

    for i in conv::inbox(&ledger, "human", None).await? {

        let from_display = conv::display_name(&ledger, &i.from).await;

        let mut v = serde_json::to_value(&i)?;

        v["from_display"] = from_display.into();

        items.push(v);

    }

    if as_json && !follow {

        println!("{}", Value::Array(items));

        return Ok(0);

    }

    if items.is_empty() && !follow {

        println!("{} {}", ui::note(), ui::dim("nothing unread"));

    }

    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();

    for i in &items {

        seen.insert(i["id"].as_str().unwrap_or("").to_string());

        if as_json {

            println!("{i}");

        } else {

            print_item(i);

        }

    }

    if !follow {

        return Ok(0);

    }

    let client = owner_client(ctx, &home).await?.context("--follow needs `spectra serve` running")?;

    if !as_json {

        println!("{} {}", ui::note(), ui::dim("following; Ctrl-C stops"));

    }

    // Every new message is a cue to look at the inbox again.

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();

    let watcher = tokio::spawn(async move {

        client

            .stream("/api/v1/events?follow=1&kind=message.posted", |l| {

                if l.starts_with("data:") {

                    let _ = tx.send(());

                }

                true

            })

            .await

    });

    while rx.recv().await.is_some() {

        for i in conv::inbox(&ledger, "human", None).await? {

            if seen.insert(i.id.clone()) {

                let mut v = serde_json::to_value(&i)?;

                v["from_display"] = conv::display_name(&ledger, &i.from).await.into();

                if as_json {

                    println!("{v}");

                } else {

                    print_item(&v);

                }

            }

        }

    }

    watcher.await??;

    Ok(0)

}

pub async fn search(

    ctx: &Ctx,

    query: &str,

    channel: Option<String>,

    author: Option<String>,

    since: Option<String>,

    limit: u32,

    as_json: bool,

) -> Result<i32> {

    let ledger = Ledger::open_read_only(&ctx.db()?).await?;

    let owner = Principals::load(&ledger).await?;

    let me = owner.owner().context("no owner is recorded")?;

    let principal = spectra_core::principals::Principal::Owner { key: me.hash(), name: me.name.clone() };

    let since = since.as_deref().and_then(|s| spectra_mcp::tools::parse_since(s, spectra_core::ids::now_ms()));

    let hits = conv::search(&ledger, &principal, query, channel.as_deref(), author.as_deref(), since, limit).await?;

    if as_json {

        println!("{}", serde_json::to_string(&hits)?);

        return Ok(0);

    }

    if hits.is_empty() {

        println!("{} {}", ui::note(), ui::dim("no matches"));

    }

    for h in hits {

        let who = conv::display_name(&ledger, &h.author).await;

        println!(

            "{} {}{}{}{}{}",

            ui::id(&h.id[..12.min(h.id.len())]),

            ui::author(who),

            ui::sep(),

            h.channel,

            ui::sep(),

            ui::dim(ui::ago(h.at_ms as u64))

        );

        println!("  {}", h.snippet);

    }

    Ok(0)

}

/// `spectra api METHOD PATH [-d JSON]`: any API call, signed as the owner,

/// JSON out (the signing proxy for scripts and the editor extensions).

pub async fn raw(ctx: &Ctx, method: &str, path: &str, data: Option<String>, owner_home_flag: Option<PathBuf>) -> Result<i32> {

    let home = owner_home(ctx, owner_home_flag);

    let c = owner_client(ctx, &home).await?.context("`spectra api` needs `spectra serve` running")?;

    let r = c.send(&method.to_uppercase(), path, &[], data.map(String::into_bytes).unwrap_or_default()).await?;

    println!("{}", String::from_utf8_lossy(&r.body));

    Ok(if r.status < 400 { 0 } else { 2 })

}

//! Every `--json` output is one JSON document on stdout and nothing else

//! (as `oxy`'s are), with and without the daemon; and `spectra say` with

//! `spectra inbox --follow` work as a terminal chat with the agents.

mod common;

use common::{add_agent, hermetic, run, setup, text, Serving};

use serde_json::{json, Value};

use spectra_mcp::client::Client;

use std::io::{BufRead, BufReader};

use std::process::Stdio;

use std::time::Duration;

fn json_of(d: &std::path::Path, args: &[&str]) -> Value {

    let o = run(d, args);

    assert!(o.status.success(), "{args:?}: {}", text(&o));

    serde_json::from_slice(&o.stdout)

        .unwrap_or_else(|e| panic!("{args:?} printed more than one JSON document ({e}): {}", String::from_utf8_lossy(&o.stdout)))

}

#[test]

fn every_json_output_parses_without_the_daemon() {

    let dir = setup();

    let d = dir.path();

    add_agent(d, "claude-oxygen", &[]);

    assert_eq!(json_of(d, &["version", "--json"])["ledger_format"], 1);

    assert_eq!(json_of(d, &["identity", "show", "--json"])["owner"]["name"], "Andy McCall");

    assert_eq!(json_of(d, &["agent", "list", "--json"])[0]["name"], "claude-oxygen");

    assert_eq!(json_of(d, &["project", "list", "--json"])[0]["name"], "oxygen");

    let posted = json_of(d, &["say", "--json", "-c", "general", "the release is tomorrow @claude-oxygen"]);

    assert_eq!(posted["channel"], "#general");

    assert!(json_of(d, &["inbox", "--json"]).as_array().unwrap().is_empty());

    assert_eq!(json_of(d, &["search", "--json", "release"]).as_array().unwrap().len(), 1);

    let v = json_of(d, &["admin", "verify", "--json"]);

    assert!(v["findings"].as_array().unwrap().is_empty(), "a CLI post verifies: {v}");

}

#[tokio::test]

async fn a_terminal_chat_with_the_agents() {

    let dir = setup();

    let d = dir.path();

    let claude = add_agent(d, "claude-oxygen", &[]);

    let srv = Serving::start(d);

    let a = Client::new(&srv.socket, Some(claude));

    // Through the running daemon: say, and the api proxy.

    assert_eq!(json_of(d, &["say", "--json", "hello from the terminal"])["channel"], "#general");

    let api: Value = serde_json::from_slice(&run(d, &["api", "GET", "/api/v1/brief"]).stdout).unwrap();

    assert_eq!(api["actor"], "human");

    let mut follow = hermetic(d).args(["inbox", "--follow", "--json"]).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap();

    let mut lines = BufReader::new(follow.stdout.take().unwrap()).lines();

    tokio::time::sleep(Duration::from_millis(500)).await;

    let say = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "say", "arguments": { "body": "@andy the build is green", "channel": "general" } } });

    assert_eq!(a.mcp("2025-11-25", &say).await.unwrap().status, 200);

    let got = tokio::task::spawn_blocking(move || lines.next()).await.unwrap().expect("a line").unwrap();

    let item: Value = serde_json::from_str(&got).unwrap();

    assert_eq!(item["from"], "agent:claude-oxygen");

    assert!(item["summary"].as_str().unwrap().contains("the build is green"));

    follow.kill().ok();

    follow.wait().ok();

    srv.stop();

}

- **Added: the human at a terminal (#12).** `spectra say [-c CHANNEL] [--reply-to ID] BODY`, `spectra inbox [--follow]`, `spectra search QUERY [--channel] [--author] [--since 7d]` and `spectra api METHOD PATH [-d JSON]`, each signing with the owner's own key: through the daemon when it is serving, else straight into the ledger. Every listing takes `--json` (one document); `say` and `inbox --follow` make a terminal chat with the agents.

impl Client {

    /// Send a signed GET and hand each line of the streamed body to `line`

    /// as it arrives (an SSE stream, say); stop when it returns false.

    pub async fn stream(&self, path: &str, mut line: impl FnMut(&str) -> bool) -> Result<u16> {

        let stream = tokio::net::UnixStream::connect(&self.socket)

            .await

            .with_context(|| format!("connecting to {} (is `spectra serve` running?)", self.socket.display()))?;

        let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;

        tokio::spawn(async move {

            let _ = conn.await;

        });

        let mut req = hyper::Request::builder().method("GET").uri(path).header("host", "spectra").header("connection", "close");

        if let Some(pair) = &self.signer {

            for (k, v) in sign_headers(pair, &format!("GET {path}"), b"") {

                req = req.header(k, v);

            }

        }

        let resp = sender.send_request(req.body(Full::new(Bytes::new()))?).await?;

        let status = resp.status().as_u16();

        let mut body = resp.into_body();

        let mut buf = String::new();

        while let Some(frame) = body.frame().await {

            if let Ok(data) = frame?.into_data() {

                buf.push_str(&String::from_utf8_lossy(&data));

                while let Some(i) = buf.find('\n') {

                    let l: String = buf.drain(..=i).collect();

                    if !line(l.trim_end()) {

                        return Ok(status);

                    }

                }

            }

        }

        Ok(status)

    }

}

- [x] #12 Human CLI — done 2026-11-02: say, inbox --follow, search, api (the signing proxy for spectra-vscode); tests/cli_json.rs parses every --json output with and without the daemon.

pub mod human;

    /// Post a message as yourself

    Say {

        /// What to say (Markdown; @name mentions)

        body: String,

        /// The channel (default: #general)

        #[arg(short, long)]

        channel: Option<String>,

        /// Reply in the thread of this message

        #[arg(long)]

        reply_to: Option<String>,

        #[arg(long)]

        json: bool,

        #[arg(long)]

        owner_home: Option<PathBuf>,

    },

    /// Your unread mentions and direct messages; --follow waits for more

    Inbox {

        #[arg(long)]

        follow: bool,

        #[arg(long)]

        json: bool,

        #[arg(long)]

        owner_home: Option<PathBuf>,

    },

    /// Search what was said: words, "phrases", prefix*, AND/OR/NOT

    Search {

        query: String,

        #[arg(long)]

        channel: Option<String>,

        #[arg(long)]

        author: Option<String>,

        /// 7d, 3h, 30m

        #[arg(long)]

        since: Option<String>,

        #[arg(long, default_value_t = 20)]

        limit: u32,

        #[arg(long)]

        json: bool,

    },

    /// Any API call, signed as the owner: JSON out

    Api {

        method: String,

        path: String,

        /// The request body (JSON)

        #[arg(short = 'd', long = "body")]

        body: Option<String>,

        #[arg(long)]

        owner_home: Option<PathBuf>,

    },

        Command::Say { body, channel, reply_to, json, owner_home } => {

            runtime()?.block_on(human::say(&ctx, channel, reply_to, body, owner_home, json))

        }

        Command::Inbox { follow, json, owner_home } => runtime()?.block_on(human::inbox(&ctx, follow, owner_home, json)),

        Command::Search { query, channel, author, since, limit, json } => {

            runtime()?.block_on(human::search(&ctx, &query, channel, author, since, limit, json))

        }

        Command::Api { method, path, body, owner_home } => runtime()?.block_on(human::raw(&ctx, &method, &path, body, owner_home)),