spectra serve and Tuner: MCP over the Unix socket and loopback, signed requests, both protocol revisions, owner events through the daemon (#7) 571572f5fb2c

//! The daemon's state (spec §7.1): the ledger it alone writes, the

//! principals kept current from the bus, the authenticator, and presence

//! with its timers. HTTP lives elsewhere (`spectra-mcp`, `radar`); this is

//! what they share.

use crate::authn::{Authenticator, Caller, Refusal};

use crate::event::{Auth, Event, Method, NewEvent};

use crate::keys::KeyPair;

use crate::ledger::{KeyResolver, Ledger};

use crate::presence::Presence;

use crate::principals::{Principal, Principals};

use crate::proof::Proof;

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

use serde::{Deserialize, Serialize};

use std::path::{Path, PathBuf};

use std::sync::Arc;

use tokio::sync::{Mutex, RwLock};

/// Kinds the owner's CLI may hand the daemon, already signed over their

/// payload (spec §8.5, "One writer").

pub const OWNER_KINDS: &[&str] = &["identity.rotated", "agent.added", "agent.disabled", "project.added"];

pub struct Daemon {

    pub data: PathBuf,

    pub ledger: Ledger,

    pub principals: RwLock<Principals>,

    pub authn: Authenticator,

    pub presence: Mutex<Presence>,

}

impl Daemon {

    /// Open the data directory's ledger for writing and start the timers.

    pub async fn start(data: &Path) -> Result<Arc<Daemon>> {

        let db = data.join("spectra.db");

        if !db.exists() {

            bail!("no Spectra in {} (run `spectra init`)", data.display());

        }

        let instance = KeyPair::instance(data, "spectra")?;

        let ledger = Ledger::open(&db, Some(instance)).await?;

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

        let d = Arc::new(Daemon {

            data: data.to_path_buf(),

            ledger,

            principals: RwLock::new(principals),

            authn: Authenticator::default(),

            presence: Mutex::new(Presence::default()),

        });

        d.clone().watch_principals();

        d.clone().tick_presence();

        Ok(d)

    }

    /// Reload who is who whenever an identity or agent event is appended.

    fn watch_principals(self: Arc<Self>) {

        let mut rx = self.ledger.subscribe();

        tokio::spawn(async move {

            loop {

                match rx.recv().await {

                    Ok(e)

                        if e.kind.starts_with("identity.")

                            || e.kind.starts_with("agent.") && e.kind != "agent.presence" && e.kind != "agent.status" =>

                    {

                        if let Ok(p) = Principals::load(&self.ledger).await {

                            *self.principals.write().await = p;

                        }

                    }

                    Ok(_) => {}

                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {

                        if let Ok(p) = Principals::load(&self.ledger).await {

                            *self.principals.write().await = p;

                        }

                    }

                    Err(_) => return,

                }

            }

        });

    }

    /// Record presence changes and quiet sessions every 15 s, and write a

    /// checkpoint when one is due by time.

    fn tick_presence(self: Arc<Self>) {

        tokio::spawn(async move {

            let mut every = tokio::time::interval(std::time::Duration::from_secs(15));

            loop {

                every.tick().await;

                let events = self.presence.lock().await.tick(crate::ids::now_ms());

                if !events.is_empty() {

                    let _ = self.ledger.append(events).await;

                }

                let _ = self.ledger.append(vec![]).await;

            }

        });

    }

    /// Authenticate a request (spec §8.5); an agent used from the wrong Unix

    /// user is recorded as `guard.impersonation`, and a first connection

    /// that binds an agent's uid as `agent.uid_bound`.

    pub async fn authenticate(&self, proof: Option<&Proof>, method: &str, body: &[u8], peer: Option<Peer>) -> Result<Caller, Refusal> {

        let now = crate::ids::now_ms();

        let r = {

            let p = self.principals.read().await;

            self.authn.authenticate(&p, proof, method, body, now, peer.map(|x| x.uid))

        };

        match &r {

            Err(Refusal::WrongUid { agent, bound, from }) => {

                let e = NewEvent::system(

                    "guard.impersonation",

                    serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": from, "from_pid": peer.map(|x| x.pid), "method": method }),

                )

                .subject(agent.clone());

                let _ = self.ledger.append_one(e).await;

                tracing::warn!(agent, bound, from, "a key used from the wrong Unix user");

            }

            Ok(Caller { principal: Principal::Agent { name, .. }, learn_uid: Some(uid), .. }) => {

                let e = NewEvent::system("agent.uid_bound", serde_json::json!({ "name": name, "uid": uid })).subject(name.clone());

                let _ = self.ledger.append_one(e).await;

            }

            _ => {}

        }

        r

    }

    /// Note a request from an agent: its session, and the event that

    /// starts one when this is the first in ten minutes.

    pub async fn touch(&self, who: &Principal, new_session: bool) -> Option<String> {

        let Principal::Agent { name, .. } = who else { return None };

        let (id, started) = self.presence.lock().await.request(name, crate::ids::now_ms(), new_session);

        if let Some(e) = started {

            let _ = self.ledger.append_one(e).await;

        }

        Some(id)

    }

    /// Append events the owner's CLI signed over their payload, after

    /// checking each as `spectra admin verify` would.

    pub async fn accept_owner_events(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {

        let mut principals = self.principals.read().await.clone();

        let projects: Vec<String> = events

            .iter()

            .filter(|e| e.kind == "project.added")

            .filter_map(|e| e.payload.get("name").and_then(|n| n.as_str()).map(String::from))

            .collect();

        let mut out = Vec::new();

        for w in events {

            let n = w.into_new_event()?;

            if n.kind == "channel.created" {

                // A project's channel comes with its project, unsigned, as `project add` makes it.

                let p = &n.payload;

                let ok = p.get("kind").and_then(|k| k.as_str()) == Some("project")

                    && p.get("name").and_then(|x| x.as_str()).is_some_and(|x| projects.iter().any(|q| q == x));

                if !ok || n.auth.method != Method::System {

                    bail!("a channel is only made with its project");

                }

                out.push(n);

                continue;

            }

            if !OWNER_KINDS.contains(&n.kind.as_str()) {

                bail!("{} cannot be handed to the daemon", n.kind);

            }

            if n.actor != "human" {

                bail!("only the owner hands events to the daemon");

            }

            // Check it as verify will, against who is who including the

            // batch so far.

            let probe = probe_event(&n);

            principals.check(&probe).map_err(|why| anyhow::anyhow!("refused {}: {why}", n.kind))?;

            if n.kind != "identity.rotated" {

                let key = oxy_core::hash::ObjectRef(n.auth.key.context("unsigned")?);

                if !matches!(principals.principal(&key, crate::ids::now_ms()), Ok(Principal::Owner { .. })) {

                    bail!("refused {}: not signed by the owner", n.kind);

                }

            }

            out.push(n);

        }

        let appended = self.ledger.append(out).await?;

        let _ = self.ledger.checkpoint().await;

        // Who is who has changed: know it before the next request arrives.

        *self.principals.write().await = Principals::load(&self.ledger).await?;

        Ok(appended.iter().map(|e| e.seq).collect())

    }

    /// Write a checkpoint covering everything (at shutdown).

    pub async fn shutdown(&self) {

        let _ = self.ledger.checkpoint().await;

    }

}

/// The Unix peer of a socket connection.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub struct Peer {

    pub uid: u32,

    pub pid: Option<i32>,

}

/// An event as the CLI hands it to the daemon: signed already.

#[derive(Clone, Debug, Serialize, Deserialize)]

pub struct WireEvent {

    pub kind: String,

    pub actor: String,

    pub subject: Option<String>,

    pub project: Option<String>,

    pub payload: serde_json::Value,

    /// Hex of the auth key and the encoded proof; absent for a project's channel.

    pub key: Option<String>,

    pub proof: Option<String>,

}

impl WireEvent {

    pub fn from_new(n: &NewEvent) -> WireEvent {

        WireEvent {

            kind: n.kind.clone(),

            actor: n.actor.clone(),

            subject: n.subject.clone(),

            project: n.project.clone(),

            payload: n.payload.clone(),

            key: n.auth.key.map(hex::encode),

            proof: n.auth.proof.as_ref().map(hex::encode),

        }

    }

    fn into_new_event(self) -> Result<NewEvent> {

        let auth = match (&self.key, &self.proof) {

            (Some(k), Some(p)) => Auth {

                key: Some(hex::decode(k).ok().and_then(|b| b.try_into().ok()).context("a key is 32 bytes of hex")?),

                method: Method::Signature,

                assurance: 2,

                proof: Some(hex::decode(p).context("a proof is hex")?),

            },

            _ => Auth::system(),

        };

        let mut n = NewEvent::system(&self.kind, self.payload).by(self.actor, auth);

        n.subject = self.subject;

        n.project = self.project;

        Ok(n)

    }

}

/// The event `n` would be, for checking before it is appended.

fn probe_event(n: &NewEvent) -> Event {

    let now = crate::ids::now_ms();

    Event {

        seq: 0,

        id: String::new(),

        at_ms: now,

        hlc: crate::hlc::Hlc { wall_ms: now, logical: 0 },

        prev: [0; 32],

        hash: [0; 32],

        auth: n.auth.clone(),

        kind: n.kind.clone(),

        actor: n.actor.clone(),

        session: n.session.clone(),

        project: n.project.clone(),

        subject: n.subject.clone(),

        payload: Some(crate::event::canonical_json(&n.payload)),

        trace: n.trace.clone(),

    }

}

//! A client for the daemon over its Unix socket (spec §8.5), signing every

//! request with the caller's key: the shims and the CLI use it, and so do

//! the tests.

use anyhow::{Context, Result};

use http_body_util::{BodyExt, Full};

use hyper::body::Bytes;

use spectra_core::keys::KeyPair;

use spectra_core::proof::Proof;

use std::path::{Path, PathBuf};

/// A response, its body read whole.

#[derive(Debug)]

pub struct Response {

    pub status: u16,

    pub headers: Vec<(String, String)>,

    pub body: Vec<u8>,

}

impl Response {

    pub fn json(&self) -> Result<serde_json::Value> {

        serde_json::from_slice(&self.body).with_context(|| format!("not JSON: {}", String::from_utf8_lossy(&self.body)))

    }

    pub fn header(&self, name: &str) -> Option<&str> {

        self.headers.iter().find(|(k, _)| k.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str())

    }

}

pub struct Client {

    socket: PathBuf,

    signer: Option<KeyPair>,

}

impl Client {

    pub fn new(socket: &Path, signer: Option<KeyPair>) -> Client {

        Client { socket: socket.to_path_buf(), signer }

    }

    /// Whether a daemon answers on the socket.

    pub async fn alive(socket: &Path) -> bool {

        tokio::net::UnixStream::connect(socket).await.is_ok()

    }

    /// Send `method path` with `body` and extra headers, signed.

    pub async fn send(&self, method: &str, path: &str, headers: &[(&str, String)], body: Vec<u8>) -> Result<Response> {

        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(method).uri(path).header("host", "spectra").header("content-type", "application/json");

        for (k, v) in headers {

            req = req.header(*k, v);

        }

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

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

                req = req.header(k, v);

            }

        }

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

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

        let headers = resp.headers().iter().map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string())).collect();

        let body = resp.into_body().collect().await?.to_bytes().to_vec();

        Ok(Response { status, headers, body })

    }

    /// A JSON-RPC request to `/mcp` with the protocol headers of `version`.

    pub async fn mcp(&self, version: &str, request: &serde_json::Value) -> Result<Response> {

        let method = request.get("method").and_then(|m| m.as_str()).unwrap_or("").to_string();

        let mut headers = vec![

            ("accept", "application/json, text/event-stream".to_string()),

            ("mcp-protocol-version", version.to_string()),

            ("mcp-method", method.clone()),

        ];

        if let Some(name) = request.pointer("/params/name").or_else(|| request.pointer("/params/uri")).and_then(|n| n.as_str()) {

            headers.push(("mcp-name", name.to_string()));

        }

        self.send("POST", "/mcp", &headers, serde_json::to_vec(request)?).await

    }

}

/// The signature headers of spec §8.5 for `method_path` and `body`.

pub fn sign_headers(pair: &KeyPair, method_path: &str, body: &[u8]) -> Vec<(&'static str, String)> {

    let p = Proof::sign(pair, method_path, body);

    vec![

        ("spectra-key", p.sig.key.to_hex()),

        ("spectra-time", p.time_ms.to_string()),

        ("spectra-nonce", hex::encode(p.nonce)),

        ("spectra-signature", hex::encode(oxy_core::cbor::encode(&p.sig.to_value()))),

    ]

}

/// The JSON-RPC result inside a response, whether it came as JSON or as

/// the last `data:` of an SSE stream.

pub fn rpc_result(r: &Response) -> Result<serde_json::Value> {

    let is_sse = r.header("content-type").is_some_and(|c| c.starts_with("text/event-stream"));

    if !is_sse {

        return r.json();

    }

    let text = String::from_utf8_lossy(&r.body);

    let last = text

        .lines()

        .rev()

        .filter_map(|l| l.strip_prefix("data:"))

        .map(str::trim)

        .find(|l| !l.is_empty())

        .context("an SSE response with no data")?;

    Ok(serde_json::from_str(last)?)

}

//! Tuner over HTTP (spec §15.1, §8.5, §21.1): `POST /mcp` on the Unix

//! socket and the loopback port, every request signed and checked before

//! anything else; the peer's uid read from the socket; `Origin` checked

//! against DNS rebinding.

use crate::protocol::{self, Headers, Reply};

use crate::tools::Ctx;

use axum::body::Bytes;

use axum::extract::connect_info::{ConnectInfo, Connected};

use axum::extract::State;

use axum::http::{HeaderMap, StatusCode};

use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};

use axum::response::{IntoResponse, Response};

use axum::routing::post;

use axum::Router;

use futures::stream::{self, StreamExt};

use serde_json::{json, Value as Json};

use spectra_core::authn::Caller;

use spectra_core::daemon::{Daemon, Peer, WireEvent};

use spectra_core::principals::Principal;

use spectra_core::proof::Proof;

use std::convert::Infallible;

use std::sync::Arc;

use std::time::Duration;

/// Who is on the other end of a connection: a Unix peer, or TCP.

#[derive(Clone, Copy, Debug)]

pub struct PeerInfo(pub Option<Peer>);

impl Connected<axum::serve::IncomingStream<'_, tokio::net::UnixListener>> for PeerInfo {

    fn connect_info(stream: axum::serve::IncomingStream<'_, tokio::net::UnixListener>) -> Self {

        let cred = stream.io().peer_cred().ok();

        PeerInfo(cred.map(|c| Peer { uid: c.uid(), pid: c.pid() }))

    }

}

impl Connected<axum::serve::IncomingStream<'_, tokio::net::TcpListener>> for PeerInfo {

    fn connect_info(_: axum::serve::IncomingStream<'_, tokio::net::TcpListener>) -> Self {

        PeerInfo(None)

    }

}

/// The app state: the daemon, and the origins a browser may send from.

#[derive(Clone)]

pub struct AppState {

    pub daemon: Arc<Daemon>,

    pub origins: Arc<Vec<String>>,

}

/// Tuner's routes; the binary adds the API and Radar beside them.

pub fn router(state: AppState) -> Router {

    Router::new()

        .route("/mcp", post(mcp).get(not_allowed).delete(not_allowed))

        .route("/api/v1/ledger/events", post(owner_events))

        .with_state(state)

}

async fn not_allowed() -> StatusCode {

    StatusCode::METHOD_NOT_ALLOWED

}

/// The proof the signature headers carry, for `method_path` and `body`.

pub fn proof_from(headers: &HeaderMap, method_path: &str, body: &[u8]) -> Option<Proof> {

    let get = |k: &str| headers.get(k).and_then(|v| v.to_str().ok());

    let key = get("spectra-key")?;

    let time_ms = get("spectra-time")?.parse().ok()?;

    let nonce: [u8; 16] = hex::decode(get("spectra-nonce")?).ok()?.try_into().ok()?;

    let sig_bytes = hex::decode(get("spectra-signature")?).ok()?;

    let sig = oxy_core::cbor::decode(&sig_bytes).ok().and_then(|v| oxy_core::identity::Signature::from_value(&v).ok())?;

    if sig.key.to_hex() != key {

        return None;

    }

    Some(Proof { sig, method: method_path.into(), body_hash: *spectra_core::event::hash_body_plain(body), time_ms, nonce })

}

/// Refuse a browser request from a page that is not ours (spec §21.1).

pub fn origin_ok(headers: &HeaderMap, allowed: &[String]) -> bool {

    match headers.get("origin").and_then(|v| v.to_str().ok()) {

        None => true,

        Some(o) => allowed.iter().any(|a| a == o),

    }

}

/// Authenticate a request, or the response that refuses it.

pub async fn authenticate(

    state: &AppState,

    peer: PeerInfo,

    headers: &HeaderMap,

    method_path: &str,

    body: &[u8],

) -> Result<Caller, Response> {

    if !origin_ok(headers, &state.origins) {

        return Err((

            StatusCode::FORBIDDEN,

            axum::Json(protocol::error(&Json::Null, protocol::INVALID_REQUEST, "this origin may not call Spectra")),

        )

            .into_response());

    }

    let proof = proof_from(headers, method_path, body);

    state.daemon.authenticate(proof.as_ref(), method_path, body, peer.0).await.map_err(|why| {

        tracing::info!(%why, "refused a request");

        // The reason concerns the caller's own key, so the caller is told it:

        // an agent whose delegation expired must be able to say so.

        (

            StatusCode::UNAUTHORIZED,

            axum::Json(protocol::error(&Json::Null, protocol::UNAUTHENTICATED, &format!("not authenticated: {why}"))),

        )

            .into_response()

    })

}

async fn mcp(State(state): State<AppState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, body: Bytes) -> Response {

    let caller = match authenticate(&state, peer, &headers, "POST /mcp", &body).await {

        Ok(c) => c,

        Err(r) => return r,

    };

    let req: Json = match serde_json::from_slice(&body) {

        Ok(v) => v,

        Err(e) => {

            return (StatusCode::BAD_REQUEST, axum::Json(protocol::error(&Json::Null, protocol::PARSE_ERROR, &format!("not JSON: {e}"))))

                .into_response()

        }

    };

    let new_session = req.pointer("/params/arguments/new_session").and_then(Json::as_bool).unwrap_or(false)

        && req.pointer("/params/name").and_then(Json::as_str) == Some("brief");

    let session = state.daemon.touch(&caller.principal, new_session).await;

    let h = Headers {

        version: headers.get("mcp-protocol-version").and_then(|v| v.to_str().ok()),

        method: headers.get("mcp-method").and_then(|v| v.to_str().ok()),

        name: headers.get("mcp-name").and_then(|v| v.to_str().ok()),

    };

    let version = h.version.unwrap_or("2025-03-26").to_string();

    let ctx = Ctx { daemon: state.daemon.clone(), caller, session };

    let meta = meta(&ctx).await;

    match protocol::handle(&ctx, &h, req, meta.clone()).await {

        Reply::Json(status, v) => (StatusCode::from_u16(status).unwrap_or(StatusCode::OK), axum::Json(v)).into_response(),

        Reply::Accepted => StatusCode::ACCEPTED.into_response(),

        Reply::Stream(id, tool, args) => {

            let fut = async move {

                let out = (tool.run)(&ctx, args).await;

                let meta = crate::http::meta(&ctx).await;

                let msg = protocol::stream_result(&id, &version, &out, meta);

                Ok::<_, Infallible>(SseEvent::default().event("message").data(msg.to_string()))

            };

            sse(stream::once(fut).boxed())

        }

        Reply::Listen(id, sub) => {

            let ack = json!({ "jsonrpc": "2.0", "method": "notifications/subscriptions/acknowledged", "params": { "_meta": { "io.modelcontextprotocol/subscriptionId": sub } } });

            let _ = id;

            let first = stream::once(async move { Ok::<_, Infallible>(SseEvent::default().event("message").data(ack.to_string())) });

            // Spectra's tool list never changes while it runs: the stream

            // stays open, quiet but for keep-alives, until the client closes it.

            sse(first.chain(stream::pending()).boxed())

        }

    }

}

fn sse(s: futures::stream::BoxStream<'static, Result<SseEvent, Infallible>>) -> Response {

    let mut r = Sse::new(s).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)).text("")).into_response();

    r.headers_mut().insert("x-accel-buffering", "no".parse().unwrap());

    r

}

/// `_meta` on every result (spec §15.2): the session and the unread count;

/// the rules version arrives with the rules (#8).

pub async fn meta(ctx: &Ctx) -> Json {

    let mut m = serde_json::Map::new();

    if let Some(s) = &ctx.session {

        m.insert("spectra/session".into(), json!(s));

    }

    if let Ok(n) = spectra_core::conversation::unread(&ctx.daemon.ledger, &ctx.caller.principal.actor()).await {

        m.insert("spectra/inbox_unread".into(), json!(n));

    }

    Json::Object(m)

}

/// `POST /api/v1/ledger/events`: the owner's CLI hands over events it

/// signed over their payload, so `agent add` and `project add` work while

/// the daemon holds the ledger (spec §8.5, "One writer").

async fn owner_events(

    State(state): State<AppState>,

    ConnectInfo(peer): ConnectInfo<PeerInfo>,

    headers: HeaderMap,

    body: Bytes,

) -> Response {

    let caller = match authenticate(&state, peer, &headers, "POST /api/v1/ledger/events", &body).await {

        Ok(c) => c,

        Err(r) => return r,

    };

    if !matches!(caller.principal, Principal::Owner { .. }) {

        return (StatusCode::FORBIDDEN, axum::Json(json!({ "title": "only the owner hands events to the daemon" }))).into_response();

    }

    let events: Vec<WireEvent> = match serde_json::from_slice(&body) {

        Ok(v) => v,

        Err(e) => return (StatusCode::BAD_REQUEST, axum::Json(json!({ "title": format!("not a list of events: {e}") }))).into_response(),

    };

    match state.daemon.accept_owner_events(events).await {

        Ok(seqs) => axum::Json(json!({ "appended": seqs })).into_response(),

        Err(e) => (StatusCode::UNPROCESSABLE_ENTITY, axum::Json(json!({ "title": e.to_string() }))).into_response(),

    }

}

//! The MCP protocol (spec §15.1): revision 2026-07-28, stateless, with its

//! mirrored headers checked against the body; and 2025-11-25, 2025-06-18

//! and 2025-03-26, which begin with `initialize`. No sessions are minted;

//! Spectra's own session (spec §8.2) rides in `_meta`.

use crate::tools::{self, Ctx, Output};

use crate::{INSTRUCTIONS, PROTOCOL_VERSIONS, STATELESS};

use serde_json::{json, Value as Json};

/// JSON-RPC and MCP error codes.

pub const PARSE_ERROR: i64 = -32700;

pub const INVALID_REQUEST: i64 = -32600;

pub const METHOD_NOT_FOUND: i64 = -32601;

pub const INVALID_PARAMS: i64 = -32602;

pub const HEADER_MISMATCH: i64 = -32020;

pub const UNSUPPORTED_VERSION: i64 = -32022;

/// Implementation-defined: the request did not authenticate (spec §8.5).

pub const UNAUTHENTICATED: i64 = -32010;

/// How a request is answered.

pub enum Reply {

    /// A JSON body with this HTTP status.

    Json(u16, Json),

    /// 202 and no body: a notification.

    Accepted,

    /// A tool that takes long: run it, answering as SSE with keep-alives.

    Stream(Json, &'static tools::Tool, Json),

    /// `subscriptions/listen`: an SSE stream that stays open.

    Listen(Json, String),

}

/// The revision a request speaks, and the headers it mirrors.

pub struct Headers<'a> {

    pub version: Option<&'a str>,

    pub method: Option<&'a str>,

    pub name: Option<&'a str>,

}

pub fn error(id: &Json, code: i64, message: &str) -> Json {

    json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })

}

fn ok(id: &Json, version: &str, mut result: Json, meta: Json) -> Json {

    if let Some(o) = result.as_object_mut() {

        if version == STATELESS {

            o.insert("resultType".into(), json!("complete"));

        }

        let mut m = meta.as_object().cloned().unwrap_or_default();

        if version == STATELESS {

            m.insert("io.modelcontextprotocol/serverInfo".into(), server_info());

        }

        if !m.is_empty() {

            o.insert("_meta".into(), Json::Object(m));

        }

    }

    json!({ "jsonrpc": "2.0", "id": id, "result": result })

}

fn server_info() -> Json {

    json!({ "name": "spectra", "version": env!("CARGO_PKG_VERSION") })

}

fn capabilities() -> Json {

    json!({ "tools": { "listChanged": false } })

}

/// Check the mirrored headers of a 2026-07-28 request against its body.

fn check_headers(h: &Headers, req: &Json) -> Result<(), String> {

    let method = req.get("method").and_then(Json::as_str).unwrap_or("");

    if h.method != Some(method) {

        return Err(format!("Mcp-Method header {:?} does not match the body's {method:?}", h.method));

    }

    let body_version = req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str);

    if body_version.is_some_and(|v| Some(v) != h.version) {

        return Err("MCP-Protocol-Version does not match the body's protocolVersion".into());

    }

    if matches!(method, "tools/call" | "resources/read" | "prompts/get") {

        let name = req.pointer("/params/name").or_else(|| req.pointer("/params/uri")).and_then(Json::as_str);

        if name.is_none() || h.name != name {

            return Err(format!("Mcp-Name header {:?} does not match the body's {name:?}", h.name));

        }

    }

    Ok(())

}

/// Answer one JSON-RPC message from an authenticated caller.

pub async fn handle(ctx: &Ctx, h: &Headers<'_>, req: Json, meta: Json) -> Reply {

    let id = req.get("id").cloned().unwrap_or(Json::Null);

    let Some(method) = req.get("method").and_then(Json::as_str).map(String::from) else {

        return Reply::Json(400, error(&id, INVALID_REQUEST, "a JSON-RPC request needs a method"));

    };

    // The revision: the header, else the legacy default of the transport.

    let version = h.version.unwrap_or("2025-03-26");

    if !PROTOCOL_VERSIONS.contains(&version) {

        let mut e = error(&id, UNSUPPORTED_VERSION, &format!("unsupported protocol version {version}"));

        e["error"]["data"] = json!({ "supported": PROTOCOL_VERSIONS });

        return Reply::Json(400, e);

    }

    if version == STATELESS {

        if let Err(why) = check_headers(h, &req) {

            return Reply::Json(400, error(&id, HEADER_MISMATCH, &why));

        }

    }

    let is_notification = req.get("id").is_none();

    match method.as_str() {

        "initialize" => {

            let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or("2025-03-26");

            let chosen = if PROTOCOL_VERSIONS[1..].contains(&asked) { asked } else { PROTOCOL_VERSIONS[1] };

            let r = json!({ "protocolVersion": chosen, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });

            Reply::Json(200, ok(&id, chosen, r, meta))

        }

        "server/discover" => {

            let r = json!({ "supportedVersions": PROTOCOL_VERSIONS, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });

            Reply::Json(200, ok(&id, version, r, meta))

        }

        "ping" => Reply::Json(200, ok(&id, version, json!({}), meta)),

        _ if is_notification => Reply::Accepted,

        "tools/list" => {

            let tools: Vec<Json> = tools::all().iter().map(|t| t.describe()).collect();

            let mut r = json!({ "tools": tools });

            if version == STATELESS {

                r["ttlMs"] = json!(3_600_000);

                r["cacheScope"] = json!("private");

            }

            Reply::Json(200, ok(&id, version, r, meta))

        }

        "tools/call" => {

            let name = req.pointer("/params/name").and_then(Json::as_str).unwrap_or("");

            let Some(tool) = tools::find(name) else {

                return Reply::Json(200, error(&id, INVALID_PARAMS, &format!("no tool named {name}")));

            };

            let args = req.pointer("/params/arguments").cloned().unwrap_or_else(|| json!({}));

            if tool.long {

                return Reply::Stream(id, tool, args);

            }

            let out = (tool.run)(ctx, args).await;

            Reply::Json(200, ok(&id, version, tool_result(&out), meta))

        }

        "subscriptions/listen" if version == STATELESS => Reply::Listen(id, spectra_core::ids::ulid()),

        _ => Reply::Json(if version == STATELESS { 404 } else { 200 }, error(&id, METHOD_NOT_FOUND, &format!("no method {method}"))),

    }

}

/// A tool's output as a `tools/call` result.

pub fn tool_result(out: &Output) -> Json {

    let mut r = json!({ "content": [{ "type": "text", "text": out.text }], "isError": out.is_error });

    if let Some(s) = &out.structured {

        r["structuredContent"] = s.clone();

    }

    r

}

/// The final message of a streamed tool call.

pub fn stream_result(id: &Json, version: &str, out: &Output, meta: Json) -> Json {

    ok(id, version, tool_result(out), meta)

}

//! The tools agents call (spec §15.2). Each is a plain function, so the

//! list is fixed and deterministic, which keeps clients' prompt caches warm.

//! A failure is a result with `isError`, never a protocol error, so the

//! model sees why.

use futures::future::BoxFuture;

use serde_json::{json, Value as Json};

use spectra_core::authn::Caller;

use spectra_core::daemon::Daemon;

use spectra_core::principals::Principal;

use std::sync::Arc;

/// What a tool sees: the daemon and who called it.

pub struct Ctx {

    pub daemon: Arc<Daemon>,

    pub caller: Caller,

    pub session: Option<String>,

}

/// A tool's result: text for the model, the same as JSON, and whether it

/// failed.

#[derive(Clone, Debug, Default)]

pub struct Output {

    pub text: String,

    pub structured: Option<Json>,

    pub is_error: bool,

}

impl Output {

    pub fn ok(text: impl Into<String>, structured: Json) -> Output {

        Output { text: text.into(), structured: Some(structured), is_error: false }

    }

    pub fn err(text: impl Into<String>) -> Output {

        Output { text: text.into(), structured: None, is_error: true }

    }

}

pub type Run = for<'a> fn(&'a Ctx, Json) -> BoxFuture<'a, Output>;

/// One tool as `tools/list` describes it.

pub struct Tool {

    pub name: &'static str,

    pub description: &'static str,

    pub schema: fn() -> Json,

    pub read_only: bool,

    pub destructive: bool,

    /// Answered as an SSE stream with keep-alives (`wait`).

    pub long: bool,

    pub run: Run,

}

impl Tool {

    pub fn describe(&self) -> Json {

        json!({

            "name": self.name,

            "description": self.description,

            "inputSchema": (self.schema)(),

            "annotations": { "readOnlyHint": self.read_only, "destructiveHint": self.destructive },

        })

    }

}

/// Every tool, in the order `tools/list` gives them.

pub fn all() -> &'static [Tool] {

    TOOLS

}

pub fn find(name: &str) -> Option<&'static Tool> {

    TOOLS.iter().find(|t| t.name == name)

}

static TOOLS: &[Tool] = &[Tool {

    name: "agents",

    description: "Every agent with its owner, kind, project, presence and status.",

    schema: || json!({ "type": "object", "properties": { "project": { "type": "string", "description": "Only this project's agents" } }, "additionalProperties": false }),

    read_only: true,

    destructive: false,

    long: false,

    run: agents,

}];

/// A principal as the model reads it.

pub fn who(p: &Principal) -> String {

    p.display()

}

fn agents(ctx: &Ctx, args: Json) -> BoxFuture<'_, Output> {

    Box::pin(async move {

        let project = args.get("project").and_then(Json::as_str);

        let rows = sqlx_rows(&ctx.daemon, project).await;

        let rows = match rows {

            Ok(r) => r,

            Err(e) => return Output::err(format!("could not list agents: {e}")),

        };

        let list: Vec<Json> = rows

            .iter()

            .map(|(name, owner, kind, project, presence, status)| {

                json!({ "name": name, "owner": owner, "kind": kind, "project": project, "presence": presence.clone().unwrap_or_else(|| "offline".into()), "status": status })

            })

            .collect();

        let text = if rows.is_empty() {

            "No agents.".to_string()

        } else {

            rows.iter()

                .map(|(name, owner, kind, project, presence, status)| {

                    let st = status.as_deref().map(|s| format!(" — \"{s}\"")).unwrap_or_default();

                    format!("- {name} · {owner}'s agent ({kind}, #{project}): {}{st}", presence.as_deref().unwrap_or("offline"))

                })

                .collect::<Vec<_>>()

                .join("\n")

        };

        Output::ok(text, json!({ "agents": list }))

    })

}

#[allow(clippy::type_complexity)]

async fn sqlx_rows(

    d: &Daemon,

    project: Option<&str>,

) -> Result<Vec<(String, String, String, String, Option<String>, Option<String>)>, sqlx::Error> {

    sqlx::query_as(

        "SELECT a.name, a.owner_name, a.kind, a.project, p.state, p.status_text FROM agents a LEFT JOIN presence p ON p.agent = a.name \

         WHERE a.disabled_ms IS NULL AND (? IS NULL OR a.project = ?) ORDER BY a.name",

    )

    .bind(project)

    .bind(project)

    .fetch_all(d.ledger.pool())

    .await

}

#!/bin/sh

# Move every crate in Cargo.lock that the OxygenFlow image does not vendor to

# the newest version of it that the image does (spectra#2), then report what

# `scripts/deps-check.sh` still finds. Run after adding a dependency.

set -eu

cd "$(dirname "$0")/.."

dev=${DEV:-..}

pkgs() { awk '/^\[\[package\]\]/{n="";v=""} /^name = /{n=$3} /^version = /{v=$3} /^source = /{print n" "v}' "$@" | tr -d '"' | sort -u; }

have=$(pkgs "$dev/oxygen/Cargo.lock" "$dev/oxygenhub/Cargo.lock" "$dev/oxygenflow/Cargo.lock")

for round in 1 2 3 4 5; do

    missing=$(pkgs Cargo.lock | grep -vxF "$have" || true)

    [ -z "$missing" ] && break

    printf '%s\n' "$missing" | while read -r name version; do

        target=$(printf '%s\n' "$have" | awk -v n="$name" '$1==n{print $2}' | sort -V | tail -1)

        [ -n "$target" ] && cargo update -q -p "$name@$version" --precise "$target" 2>/dev/null || true

    done

done

scripts/deps-check.sh

//! `spectra serve`: the daemon (spec §7.1, §21.1). It holds the ledger,

//! serves Tuner (and, as they land, the API and Radar) on the Unix socket

//! and on a loopback port, and writes a checkpoint when it stops.

use super::{ui, Ctx};

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

use spectra_core::daemon::Daemon;

use spectra_mcp::client::Client;

use spectra_mcp::http::{AppState, PeerInfo};

use std::net::SocketAddr;

use std::sync::Arc;

pub async fn serve(ctx: &Ctx, bind: &str, insecure: bool) -> Result<i32> {

    let addr: SocketAddr = bind.parse().with_context(|| format!("--bind {bind} is not an address and port"))?;

    if !addr.ip().is_loopback() && !insecure {

        bail!("Spectra serves loopback only unless it has TLS (spec §21.1); pass --insecure to bind {addr} in plain HTTP anyway");

    }

    let socket = ctx.socket();

    if socket.exists() {

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

            bail!("Spectra is already serving {} (its socket answers)", ctx.data.display());

        }

        std::fs::remove_file(&socket).ok();

    }

    let daemon = Daemon::start(&ctx.data).await?;

    let unix = tokio::net::UnixListener::bind(&socket).with_context(|| format!("binding {}", socket.display()))?;

    #[cfg(unix)]

    {

        use std::os::unix::fs::PermissionsExt;

        std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o660))?;

    }

    let tcp = tokio::net::TcpListener::bind(addr).await.with_context(|| format!("binding {addr}"))?;

    let port = tcp.local_addr()?.port();

    let origins = vec![format!("http://127.0.0.1:{port}"), format!("http://localhost:{port}"), format!("http://{addr}")];

    let state = AppState { daemon: daemon.clone(), origins: Arc::new(origins) };

    let app = spectra_mcp::http::router(state);

    let (head, _) = daemon.ledger.head().await;

    let agents = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM agents WHERE disabled_ms IS NULL")

        .fetch_one(daemon.ledger.pool())

        .await

        .unwrap_or(0);

    println!("{} {}", ui::done("Serving"), ui::subject("Spectra"));

    println!("{}", ui::kv("socket", ui::path(socket.display())));

    println!("{}", ui::kv("http", format!("http://127.0.0.1:{port}")));

    println!("{}", ui::kv("ledger", format!("{head} events{}{agents} agent(s)", ui::sep())));

    println!("  {} {}", ui::note(), ui::dim("agents connect with: spectra mcp --agent NAME"));

    let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);

    let stop = |mut rx: tokio::sync::watch::Receiver<bool>| async move {

        let _ = rx.wait_for(|s| *s).await;

    };

    let unix_srv =

        axum::serve(unix, app.clone().into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx.clone()));

    let tcp_srv = axum::serve(tcp, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx));

    let signal = async move {

        shutdown_signal().await;

        let _ = stop_tx.send(true);

    };

    let (u, t, _) = tokio::join!(unix_srv, tcp_srv, signal);

    u?;

    t?;

    daemon.shutdown().await;

    std::fs::remove_file(&socket).ok();

    println!("{} {}", ui::done("Stopped"), ui::dim(format!("ledger checkpointed at {} events", daemon.ledger.head().await.0)));

    Ok(0)

}

async fn shutdown_signal() {

    #[cfg(unix)]

    {

        let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("a SIGTERM handler");

        tokio::select! {

            _ = tokio::signal::ctrl_c() => {}

            _ = term.recv() => {}

        }

    }

    #[cfg(not(unix))]

    let _ = tokio::signal::ctrl_c().await;

}

//! `spectra serve` end to end (spec §8.5, §15.1): the real binary, a client

//! signing requests over the Unix socket, both MCP revisions, and every

//! refusal the spec promises. Hermetic: temporary homes, data and port 0.

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

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

use serde_json::json;

use spectra_core::keys::KeyPair;

use spectra_core::principals::DELEGATION_SCOPE;

use spectra_mcp::client::{rpc_result, Client};

use std::path::{Path, PathBuf};

use std::process::{Child, Command, Output};

fn hermetic(dir: &Path) -> Command {

    let mut c = Command::new(env!("CARGO_BIN_EXE_spectra"));

    c.env("SPECTRA_DATA", dir.join("data"))

        .env("OXY_HOME", dir.join("owner"))

        .env("HOME", dir)

        .env("XDG_DATA_HOME", dir.join("xdg"))

        .env("NO_COLOR", "1");

    c

}

fn run(dir: &Path, args: &[&str]) -> Output {

    hermetic(dir).args(args).output().unwrap()

}

fn text(o: &Output) -> String {

    format!("{}{}", String::from_utf8_lossy(&o.stdout), String::from_utf8_lossy(&o.stderr))

}

/// `oxy identity delegate add` in the owner's home.

fn delegate(owner_home: &Path, agent: &KeyPair) {

    let cur = KeyPair::load_home(owner_home).unwrap();

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

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

    let mut d = cur.identity.delegates.clone();

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

    let next = cur.identity.successor(d, None, spectra_core::ids::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(&owner_home.join("config.toml"), "author.key", &k.to_hex()).unwrap();

}

/// Make an agent's home, delegate to it, and add it.

fn add_agent(d: &Path, name: &str, extra: &[&str]) -> KeyPair {

    let home = d.join("agents").join(name);

    let pair = KeyPair::create_home(&home, name, "").unwrap();

    delegate(&d.join("owner"), &pair);

    let mut args =

        vec!["agent", "add", name, "--kind", "claude-code", "--project", "oxygen", "--home", home.to_str().unwrap(), "--wait", "0"];

    args.extend_from_slice(extra);

    let o = run(d, &args);

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

    pair

}

struct Serving {

    child: Child,

    socket: PathBuf,

}

impl Serving {

    fn start(d: &Path) -> Serving {

        let mut child = hermetic(d).args(["serve", "--bind", "127.0.0.1:0"]).stdout(std::process::Stdio::piped()).spawn().unwrap();

        let socket = d.join("data/spectra.sock");

        for _ in 0..200 {

            if std::os::unix::net::UnixStream::connect(&socket).is_ok() {

                return Serving { child, socket };

            }

            std::thread::sleep(std::time::Duration::from_millis(25));

        }

        child.kill().ok();

        child.wait().ok();

        panic!("spectra serve did not start");

    }

    fn stop(mut self) {

        unsafe { libc::kill(self.child.id() as i32, libc::SIGTERM) };

        let status = self.child.wait().unwrap();

        assert!(status.success(), "a clean stop on SIGTERM");

        assert!(!self.socket.exists(), "the socket is removed");

    }

}

fn setup() -> tempfile::TempDir {

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

    let d = dir.path();

    KeyPair::create_home(&d.join("owner"), "Andy McCall", "").unwrap();

    assert!(run(d, &["init"]).status.success());

    std::fs::create_dir_all(d.join("oxygen")).unwrap();

    assert!(run(d, &["project", "add", d.join("oxygen").to_str().unwrap()]).status.success());

    dir

}

fn call(id: u64, name: &str, args: serde_json::Value) -> serde_json::Value {

    json!({ "jsonrpc": "2.0", "id": id, "method": "tools/call", "params": { "name": name, "arguments": args,

        "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28" } } })

}

#[tokio::test]

async fn an_agent_discovers_lists_and_calls_tools_over_the_socket() {

    let dir = setup();

    let d = dir.path();

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

    let srv = Serving::start(d);

    let c = Client::new(&srv.socket, Some(agent.clone()));

    let r = c.mcp("2026-07-28", &json!({ "jsonrpc": "2.0", "id": 1, "method": "server/discover", "params": {} })).await.unwrap();

    assert_eq!(r.status, 200, "{}", String::from_utf8_lossy(&r.body));

    let v = r.json().unwrap();

    assert_eq!(v["result"]["serverInfo"]["name"], "spectra");

    assert_eq!(v["result"]["resultType"], "complete");

    assert!(v["result"]["supportedVersions"].as_array().unwrap().contains(&json!("2025-11-25")));

    let v = c.mcp("2026-07-28", &json!({ "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} })).await.unwrap().json().unwrap();

    assert!(v["result"]["tools"].as_array().unwrap().iter().any(|t| t["name"] == "agents"));

    assert_eq!(v["result"]["ttlMs"], 3_600_000);

    assert!(v["result"]["_meta"]["spectra/session"].is_string(), "{v}");

    let r = c.mcp("2026-07-28", &call(3, "agents", json!({}))).await.unwrap();

    let v = rpc_result(&r).unwrap();

    assert_eq!(v["result"]["isError"], false, "{v}");

    assert_eq!(v["result"]["structuredContent"]["agents"][0]["owner"], "Andy McCall");

    assert!(v["result"]["content"][0]["text"].as_str().unwrap().contains("claude-oxygen · Andy McCall's agent"));

    // The legacy revisions begin with initialize.

    let v = c.mcp("2025-11-25", &json!({ "jsonrpc": "2.0", "id": 4, "method": "initialize", "params": { "protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": { "name": "t", "version": "1" } } })).await.unwrap().json().unwrap();

    assert_eq!(v["result"]["protocolVersion"], "2025-11-25");

    assert!(v["result"]["instructions"].as_str().unwrap().contains("Call `brief` first"));

    assert_eq!(c.mcp("2025-11-25", &json!({ "jsonrpc": "2.0", "method": "notifications/initialized" })).await.unwrap().status, 202);

    srv.stop();

}

#[tokio::test]

async fn every_refusal_the_spec_promises() {

    let dir = setup();

    let d = dir.path();

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

    let uid = unsafe { libc::getuid() };

    let other_uid = (uid + 1).to_string();

    let bound = add_agent(d, "bound", &["--uid", &other_uid]);

    let srv = Serving::start(d);

    let list = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": { "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28" } } });

    let unsigned = Client::new(&srv.socket, None).mcp("2026-07-28", &list).await.unwrap();

    assert_eq!(unsigned.status, 401);

    let stranger = Client::new(&srv.socket, Some(KeyPair::generate("stranger", ""))).mcp("2026-07-28", &list).await.unwrap();

    assert_eq!(stranger.status, 401);

    assert!(String::from_utf8_lossy(&stranger.body).contains("unknown key"));

    let c = Client::new(&srv.socket, Some(agent));

    let r = c

        .send(

            "POST",

            "/mcp",

            &[("mcp-protocol-version", "2026-07-28".into()), ("mcp-method", "tools/call".into())],

            serde_json::to_vec(&list).unwrap(),

        )

        .await

        .unwrap();

    assert_eq!((r.status, r.json().unwrap()["error"]["code"].as_i64()), (400, Some(-32020)), "the Mcp-Method header must match the body");

    let r = c.send("POST", "/mcp", &[("mcp-protocol-version", "1999-01-01".into())], serde_json::to_vec(&list).unwrap()).await.unwrap();

    assert_eq!((r.status, r.json().unwrap()["error"]["code"].as_i64()), (400, Some(-32022)));

    let r = c

        .send(

            "POST",

            "/mcp",

            &[("origin", "http://evil.example".into()), ("mcp-protocol-version", "2026-07-28".into()), ("mcp-method", "tools/list".into())],

            serde_json::to_vec(&list).unwrap(),

        )

        .await

        .unwrap();

    assert_eq!(r.status, 403, "a page from another origin");

    let r = c.send("GET", "/mcp", &[], vec![]).await.unwrap();

    assert_eq!(r.status, 405);

    let wrong = Client::new(&srv.socket, Some(bound)).mcp("2026-07-28", &list).await.unwrap();

    assert_eq!(wrong.status, 401);

    assert!(String::from_utf8_lossy(&wrong.body).contains(&format!("bound to uid {other_uid}")));

    srv.stop();

    let v: serde_json::Value = serde_json::from_slice(&run(d, &["admin", "verify", "--json"]).stdout).unwrap();

    assert!(v["findings"].as_array().unwrap().is_empty(), "{v}");

    assert_eq!(v["uncovered"], 0, "the stop wrote a checkpoint");

    let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", d.join("data/spectra.db").display())).await.unwrap();

    let logged: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE kind = 'guard.impersonation'").fetch_one(&pool).await.unwrap();

    assert_eq!(logged, 1, "the wrong-uid attempt is in the ledger");

}

#[tokio::test]

async fn owner_commands_go_through_the_running_daemon_and_learn_uid_binds() {

    let dir = setup();

    let d = dir.path();

    let srv = Serving::start(d);

    // While serve holds the ledger, the CLI's writes go through it.

    let learner = add_agent(d, "learner", &["--learn-uid"]);

    std::fs::create_dir_all(d.join("hub")).unwrap();

    let o = run(d, &["project", "add", d.join("hub").to_str().unwrap()]);

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

    let o = run(d, &["agent", "list", "--json"]);

    assert!(text(&o).contains("learner"), "{}", text(&o));

    let list = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} });

    assert_eq!(Client::new(&srv.socket, Some(learner)).mcp("2025-11-25", &list).await.unwrap().status, 200);

    srv.stop();

    let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", d.join("data/spectra.db").display())).await.unwrap();

    let uid: Option<i64> = sqlx::query_scalar("SELECT uid FROM agents WHERE name = 'learner'").fetch_one(&pool).await.unwrap();

    assert_eq!(uid, Some(unsafe { libc::getuid() } as i64), "bound to the uid of its first signed connection");

    let v: serde_json::Value = serde_json::from_slice(&run(d, &["admin", "verify", "--json"]).stdout).unwrap();

    assert!(v["findings"].as_array().unwrap().is_empty(), "{v}");

}

- **Added: `spectra serve` and Tuner, MCP over the Unix socket and loopback (#7).** The daemon holds the ledger and serves `POST /mcp` on `spectra.sock` (reading the caller's uid) and on `127.0.0.1:7350`. Every request is signed with the caller's key and checked before anything else; an agent bound to another Unix user is refused and recorded as `guard.impersonation`, and `--learn-uid` binds an agent on its first connection. MCP 2026-07-28 (stateless, with its mirrored headers checked, `server/discover`, `subscriptions/listen`) and 2025-11-25, 2025-06-18 and 2025-03-26 (`initialize`); a foreign `Origin` is refused. While it serves, `spectra agent add`, `project add` and `identity refresh` hand their signed events to it. It writes a checkpoint when it stops.

name = "atomic-waker"

version = "1.1.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"

[[package]]

[[package]]

name = "axum"

version = "0.8.9"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"

dependencies = [

 "axum-core",

 "axum-macros",

 "bytes",

 "form_urlencoded",

 "futures-util",

 "http",

 "http-body",

 "http-body-util",

 "hyper",

 "hyper-util",

 "itoa",

 "matchit",

 "memchr",

 "mime",

 "percent-encoding",

 "pin-project-lite",

 "serde_core",

 "serde_json",

 "serde_path_to_error",

 "serde_urlencoded",

 "sync_wrapper",

 "tokio",

 "tower",

 "tower-layer",

 "tower-service",

 "tracing",

]

[[package]]

name = "axum-core"

version = "0.5.6"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"

dependencies = [

 "bytes",

 "futures-core",

 "http",

 "http-body",

 "http-body-util",

 "mime",

 "pin-project-lite",

 "sync_wrapper",

 "tower-layer",

 "tower-service",

 "tracing",

]

[[package]]

name = "axum-macros"

version = "0.5.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 2.0.119",

]

]

[[package]]

name = "futures"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3"

dependencies = [

 "futures-channel",

 "futures-core",

 "futures-executor",

 "futures-io",

 "futures-sink",

 "futures-task",

 "futures-util",

[[package]]

name = "futures-macro"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 3.0.6",

]

 "futures-channel",

 "futures-macro",