lib.rs 83.1%

343 of 413 lines run, 54 of 148 functions · run #17 on main · run missed
//! Radar (spec §17): the human's pages, rendered with askama, refreshed
//! from the event stream, working without JavaScript. Linked by `spectra`
//! and, later, mounted by SpectraHub.
//!
//! The owner signs in with a one-time link their CLI signed (`spectra
//! open`); everything they do here is recorded with that signed request as
//! its proof (spec §8.3). Every unsafe request carries the session's CSRF
//! token and a same-origin `Origin`.

use askama::Template;
use axum::body::Bytes;
use axum::extract::connect_info::ConnectInfo;
use axum::extract::{Form, Path, Query, State};
use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
use axum::response::{Html, IntoResponse, Redirect, Response};
use axum::routing::{get, post};
use axum::Router;
use futures::StreamExt;
use serde::Deserialize;
use spectra_core::conversation::{self as conv, Author, Post, Signer};
use spectra_core::event::{Auth, Method};
use spectra_core::ids::{format_time, now_ms};
use spectra_core::principals::{Principal, Principals};
use spectra_mcp::http::{AppState, PeerInfo};
use spectra_mcp::session::{self, Session};
use std::time::Duration;

const MARK: &str = include_str!("../assets/spectra.svg");
const STYLE: &str = include_str!("../assets/style.css");
const SCRIPT: &str = include_str!("../assets/radar.js");
const COOKIE: &str = "spectra_session";

#[derive(Clone)]
pub struct RadarState {
    pub app: AppState,
}

impl RadarState {
    fn sessions(&self) -> &session::Sessions {
        &self.app.sessions
    }
}

/// Radar's routes.
pub fn router(app: AppState) -> Router {
    let state = RadarState { app };
    Router::new()
        .route("/", get(live))
        .route("/channels", get(channels))
        .route("/c/{name}", get(channel).post(post_channel))
        .route("/c/{name}/t/{id}", get(thread).post(post_thread))
        .route("/log", get(log))
        .route("/log/verify", post(verify))
        .route("/radar/link", post(link))
        .route("/signin", get(signin))
        .route("/signout", post(signout))
        .route("/radar/events", get(events))
        .route("/radar/style.css", get(|| async { asset("text/css; charset=utf-8", STYLE) }))
        .route("/radar/radar.js", get(|| async { asset("text/javascript; charset=utf-8", SCRIPT) }))
        .route("/radar/spectra.svg", get(|| async { asset("image/svg+xml", MARK) }))
        .layer(axum::middleware::map_response(security_headers))
        .with_state(state)
}

fn asset(kind: &'static str, body: &'static str) -> Response {
    ([(header::CONTENT_TYPE, kind), (header::CACHE_CONTROL, "max-age=300")], body).into_response()
}

/// CSP and friends on every response (spec §21.2).
async fn security_headers(mut r: Response) -> Response {
    let h = r.headers_mut();
    h.insert("content-security-policy", HeaderValue::from_static("default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; frame-ancestors 'none'; form-action 'self'; base-uri 'none'"));
    h.insert("x-content-type-options", HeaderValue::from_static("nosniff"));
    // `same-origin`: a form post still carries our Origin (no-referrer makes it `null`).
    h.insert("referrer-policy", HeaderValue::from_static("same-origin"));
    h.insert("x-frame-options", HeaderValue::from_static("DENY"));
    r
}

/// What every page's frame shows.
pub struct Shell {
    pub page: &'static str,
    pub owner: String,
    pub head: u64,
    pub version: &'static str,
    pub mark: &'static str,
}

/// The signed-in owner, or the page that says how to sign in.
struct Owner {
    session: Session,
    principal: Principal,
    shell: Shell,
}

impl Owner {
    fn author(&self) -> Author<'static> {
        Author {
            principal: self.principal.clone(),
            signer: Signer::Request(Auth {
                key: Some(self.session.key),
                method: Method::Session,
                assurance: 2,
                proof: Some(self.session.proof.clone()),
            }),
        }
    }
}

fn cookie(headers: &HeaderMap) -> Option<String> {
    let c = headers.get(header::COOKIE)?.to_str().ok()?;
    c.split(';').map(str::trim).find_map(|kv| kv.strip_prefix(&format!("{COOKIE}=")).map(String::from))
}

async fn owner(st: &RadarState, headers: &HeaderMap, page: &'static str) -> Result<Owner, Response> {
    let now = now_ms();
    let session = cookie(headers)
        .and_then(|id| st.sessions().get(&id, now))
        .ok_or_else(|| signin_page("Radar shows Spectra to its owner only.", StatusCode::UNAUTHORIZED))?;
    let d = &st.app.daemon;
    let principal = d
        .principals
        .read()
        .await
        .principal(&oxy_core_key(session.key), now)
        .map_err(|why| signin_page(&format!("This session no longer belongs to the owner: {why}."), StatusCode::UNAUTHORIZED))?;
    if !matches!(principal, Principal::Owner { .. }) {
        return Err(signin_page("Radar shows Spectra to its owner only.", StatusCode::FORBIDDEN));
    }
    let shell = Shell { page, owner: principal.display(), head: d.ledger.head().await.0, version: env!("CARGO_PKG_VERSION"), mark: MARK };
    Ok(Owner { session, principal, shell })
}

fn oxy_core_key(k: [u8; 32]) -> oxy_core::hash::KeyId {
    oxy_core::hash::ObjectRef(k)
}

#[derive(Template)]
#[template(path = "signin.html")]
struct SigninPage<'a> {
    mark: &'static str,
    message: &'a str,
}

fn signin_page(message: &str, status: StatusCode) -> Response {
    (status, render(&SigninPage { mark: MARK, message })).into_response()
}

fn render<T: Template>(t: &T) -> Response {
    match t.render() {
        Ok(html) => Html(html).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("could not render the page: {e}")).into_response(),
    }
}

/// Unsafe requests: the session's CSRF token and our own origin.
/// The refusal, if the form may not go through.
fn check_form(st: &RadarState, o: &Owner, headers: &HeaderMap, csrf: &str) -> Option<Response> {
    if !spectra_mcp::http::origin_ok(headers, &st.app.origins) {
        return Some((StatusCode::FORBIDDEN, "this origin may not change Spectra").into_response());
    }
    if csrf != o.session.csrf {
        return Some((StatusCode::FORBIDDEN, "the form is stale: reload the page").into_response());
    }
    None
}

/// Markdown, sanitised: what a message's body becomes on a page.
fn markdown(body: &str) -> String {
    let mut html = String::new();
    pulldown_cmark::html::push_html(
        &mut html,
        pulldown_cmark::Parser::new_ext(body, pulldown_cmark::Options::ENABLE_STRIKETHROUGH | pulldown_cmark::Options::ENABLE_TABLES),
    );
    ammonia::clean(&html)
}

// ---- Live ----

struct AgentCard {
    name: String,
    owner: String,
    kind: String,
    project: String,
    presence: String,
    status: Option<String>,
    last_tool: Option<String>,
    last_at: String,
}

struct EventRow {
    seq: i64,
    at: String,
    kind: String,
    actor: String,
    subject: Option<String>,
}

#[derive(Template)]
#[template(path = "live.html")]
struct LivePage {
    shell: Shell,
    agents: Vec<AgentCard>,
    unread: i64,
    events: Vec<EventRow>,
}

async fn live(State(st): State<RadarState>, headers: HeaderMap) -> Response {
    let o = match owner(&st, &headers, "live").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    let d = &st.app.daemon;
    let pool = d.ledger.pool();
    let rows: Vec<(String, String, String, String, Option<String>)> =
        sqlx::query_as("SELECT a.name, a.owner_name, a.kind, a.project, p.status_text FROM agents a LEFT JOIN presence p ON p.agent = a.name WHERE a.disabled_ms IS NULL ORDER BY a.name")
            .fetch_all(pool)
            .await
            .unwrap_or_default();
    let mut agents = Vec::new();
    {
        let now = now_ms();
        for (name, owner, kind, project, status) in rows {
            let last: Option<(String, i64)> =
                sqlx::query_as("SELECT tool, at_ms FROM tool_calls WHERE agent = ? ORDER BY seq DESC LIMIT 1")
                    .bind(&name)
                    .fetch_optional(pool)
                    .await
                    .ok()
                    .flatten();
            let presence = d.presence.lock().await.state(&name, now).to_string();
            agents.push(AgentCard {
                name,
                owner,
                kind,
                project,
                presence,
                status,
                last_tool: last.as_ref().map(|(t, _)| t.clone()),
                last_at: last.map(|(_, at)| format_time(at as u64)).unwrap_or_default(),
            });
        }
    }
    let unread = conv::unread(&d.ledger, "human").await.unwrap_or(0);
    let events = recent(&st, None, None, 50)
        .await
        .into_iter()
        .map(|e| EventRow { seq: e.seq, at: e.at, kind: e.kind, actor: e.actor, subject: e.subject })
        .collect();
    render(&LivePage { shell: o.shell, agents, unread, events })
}

// ---- Channels ----

struct ChannelRow {
    name: String,
    display: String,
    kind: String,
    purpose: String,
    count: i64,
    last: String,
}

/// A channel with its message count and last message time.
type ChannelCount = (String, String, String, Option<String>, i64, Option<i64>);

#[derive(Template)]
#[template(path = "channels.html")]
struct ChannelsPage {
    shell: Shell,
    channels: Vec<ChannelRow>,
}

async fn channels(State(st): State<RadarState>, headers: HeaderMap) -> Response {
    let o = match owner(&st, &headers, "channels").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    let rows: Vec<ChannelCount> = sqlx::query_as(
        "SELECT c.id, c.name, c.kind, c.purpose, (SELECT COUNT(*) FROM messages m WHERE m.channel = c.id), (SELECT MAX(at_ms) FROM messages m WHERE m.channel = c.id) \
         FROM channels c ORDER BY c.kind = 'direct', c.kind = 'system', c.name",
    )
    .fetch_all(st.app.daemon.ledger.pool())
    .await
    .unwrap_or_default();
    let channels = rows
        .into_iter()
        .map(|(_, name, kind, purpose, count, last)| ChannelRow {
            display: match name.strip_prefix("dm:") {
                Some(p) => format!("@{p}"),
                None => format!("#{name}"),
            },
            name,
            kind,
            purpose: purpose.unwrap_or_default(),
            count,
            last: last.map(|t| format_time(t as u64)).unwrap_or_default(),
        })
        .collect();
    render(&ChannelsPage { shell: o.shell, channels })
}

// ---- A channel and a thread ----

struct MessageRow {
    id: String,
    author: String,
    at: String,
    kind: String,
    replies: i64,
    redacted: bool,
    body_html: String,
}

#[derive(Template)]
#[template(path = "channel.html")]
struct ChannelPage {
    shell: Shell,
    name: String,
    display: String,
    purpose: Option<String>,
    thread: Option<String>,
    messages: Vec<MessageRow>,
    can_post: bool,
    action: String,
    csrf: String,
}

async fn show(st: &RadarState, o: Owner, name: &str, thread: Option<&str>) -> Response {
    let d = &st.app.daemon;
    let (ch, msgs) = match conv::read(&d.ledger, &o.principal, if thread.is_some() { None } else { Some(name) }, thread, None, 200).await {
        Ok(r) => r,
        Err(e) => return (StatusCode::NOT_FOUND, format!("{e:#}")).into_response(),
    };
    let mut messages = Vec::new();
    for m in msgs {
        messages.push(MessageRow {
            author: spectra_core::conversation::display_name(&d.ledger, &m.author).await,
            at: format_time(m.at_ms as u64),
            redacted: m.redacted.is_some(),
            body_html: if m.redacted.is_some() { ammonia::clean_text(&m.body) } else { markdown(&m.body) },
            id: m.id,
            kind: m.kind,
            replies: m.replies,
        });
    }
    let can_post = conv::can_post(&o.principal, &ch).is_ok();
    let action = match thread {
        Some(t) => format!("/c/{}/t/{t}", ch.name),
        None => format!("/c/{}", ch.name),
    };
    let csrf = o.session.csrf.clone();
    render(&ChannelPage {
        shell: o.shell,
        name: ch.name.clone(),
        display: ch.display(),
        purpose: ch.purpose.clone(),
        thread: thread.map(String::from),
        messages,
        can_post,
        action,
        csrf,
    })
}

async fn channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>) -> Response {
    match owner(&st, &headers, "channels").await {
        Ok(o) => show(&st, o, &name, None).await,
        Err(r) => r,
    }
}

async fn thread(State(st): State<RadarState>, headers: HeaderMap, Path((name, id)): Path<(String, String)>) -> Response {
    match owner(&st, &headers, "channels").await {
        Ok(o) => show(&st, o, &name, Some(&id)).await,
        Err(r) => r,
    }
}

#[derive(Deserialize)]
struct PostForm {
    _csrf: String,
    body: String,
    reply_to: Option<String>,
}

async fn post_to(st: &RadarState, headers: &HeaderMap, name: &str, form: PostForm, back: String) -> Response {
    let o = match owner(st, headers, "channels").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    if let Some(r) = check_form(st, &o, headers, &form._csrf) {
        return r;
    }
    let post = Post {
        channel: if form.reply_to.is_some() { None } else { Some(name.to_string()) },
        body: form.body,
        reply_to: form.reply_to,
        ..Default::default()
    };
    match conv::post(&st.app.daemon.ledger, &o.author(), post, now_ms()).await {
        Ok(_) => Redirect::to(&back).into_response(),
        Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
    }
}

async fn post_channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>, Form(form): Form<PostForm>) -> Response {
    let back = format!("/c/{name}");
    post_to(&st, &headers, &name, form, back).await
}

async fn post_thread(
    State(st): State<RadarState>,
    headers: HeaderMap,
    Path((name, id)): Path<(String, String)>,
    Form(mut form): Form<PostForm>,
) -> Response {
    form.reply_to = Some(id.clone());
    let back = format!("/c/{name}/t/{id}");
    post_to(&st, &headers, &name, form, back).await
}

// ---- The log ----

struct LogRow {
    seq: i64,
    id: String,
    at: String,
    kind: String,
    actor: String,
    subject: Option<String>,
    method: String,
    assurance: i64,
    hash: String,
    prev: String,
    key: String,
    proof: String,
    payload: String,
}

async fn recent(st: &RadarState, kind: Option<&str>, before: Option<i64>, limit: i64) -> Vec<LogRow> {
    #[allow(clippy::type_complexity)]
    let rows: Vec<(
        i64,
        String,
        i64,
        String,
        String,
        Option<String>,
        i64,
        i64,
        Vec<u8>,
        Vec<u8>,
        Option<Vec<u8>>,
        Option<Vec<u8>>,
        Option<Vec<u8>>,
    )> = sqlx::query_as(
        "SELECT seq, id, at_ms, kind, actor, subject, auth_method, auth_assurance, hash, prev, auth_key, auth_proof, payload FROM events \
         WHERE (? IS NULL OR kind = ?) AND (? IS NULL OR seq < ?) ORDER BY seq DESC LIMIT ?",
    )
    .bind(kind)
    .bind(kind)
    .bind(before)
    .bind(before)
    .bind(limit)
    .fetch_all(st.app.daemon.ledger.pool())
    .await
    .unwrap_or_default();
    let mut out = Vec::new();
    for (seq, id, at, kind, actor, subject, method, assurance, hash, prev, key, proof, payload) in rows {
        let payload = payload
            .and_then(|p| serde_json::from_slice::<serde_json::Value>(&p).ok())
            .map(|v| serde_json::to_string_pretty(&v).unwrap_or_default())
            .unwrap_or_else(|| "(pruned to its stub)".into());
        out.push(LogRow {
            seq,
            id,
            at: format_time(at as u64),
            kind,
            actor: spectra_core::conversation::display_name(&st.app.daemon.ledger, &actor).await,
            subject,
            method: Method::from_u8(method as u8).map(|m| m.name()).unwrap_or("?").to_string(),
            assurance,
            hash: hex::encode(hash),
            prev: hex::encode(prev),
            key: key.map(hex::encode).unwrap_or_else(|| "none (Spectra itself; covered by the next checkpoint)".into()),
            proof: proof.map(|p| format!("{} bytes", p.len())).unwrap_or_else(|| "none".into()),
            payload,
        });
    }
    out
}

#[derive(Template)]
#[template(path = "log.html")]
struct LogPage {
    shell: Shell,
    head_hash: String,
    verified: String,
    csrf: String,
    kinds: Vec<String>,
    kind: String,
    events: Vec<LogRow>,
    older: Option<i64>,
}

#[derive(Deserialize)]
struct LogQuery {
    kind: Option<String>,
    before: Option<i64>,
    around: Option<i64>,
}

async fn log(State(st): State<RadarState>, headers: HeaderMap, Query(q): Query<LogQuery>) -> Response {
    let o = match owner(&st, &headers, "log").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    let d = &st.app.daemon;
    let (_, head) = d.ledger.head().await;
    let last: Option<(Vec<u8>, i64)> =
        sqlx::query_as("SELECT payload, at_ms FROM events WHERE kind = 'ledger.verified' ORDER BY seq DESC LIMIT 1")
            .fetch_optional(d.ledger.pool())
            .await
            .ok()
            .flatten();
    let verified = match last {
        Some((p, at)) => {
            let v: serde_json::Value = serde_json::from_slice(&p).unwrap_or_default();
            let n = v["findings"].as_u64().unwrap_or(0);
            format!(
                "{} at {} over {} events",
                if n == 0 { "clean".to_string() } else { format!("{n} finding(s)") },
                format_time(at as u64),
                v["events"].as_u64().unwrap_or(0)
            )
        }
        None => "not yet".into(),
    };
    let kinds: Vec<String> =
        sqlx::query_scalar("SELECT kind FROM event_counts ORDER BY n DESC LIMIT 12").fetch_all(d.ledger.pool()).await.unwrap_or_default();
    let kind = q.kind.unwrap_or_default();
    let before = q.before.or(q.around.map(|a| a + 25));
    let events = recent(&st, (!kind.is_empty()).then_some(kind.as_str()), before, 100).await;
    let older = (events.len() == 100).then(|| events.last().map(|e| e.seq)).flatten();
    let csrf = o.session.csrf.clone();
    render(&LogPage { shell: o.shell, head_hash: hex::encode(head), verified, csrf, kinds, kind, events, older })
}

#[derive(Deserialize)]
struct CsrfForm {
    _csrf: String,
}

/// Verify the ledger now and record the result (`ledger.verified`).
async fn verify(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response {
    let o = match owner(&st, &headers, "log").await {
        Ok(o) => o,
        Err(r) => return r,
    };
    if let Some(r) = check_form(&st, &o, &headers, &f._csrf) {
        return r;
    }
    let ledger = &st.app.daemon.ledger;
    match ledger.verify(&mut Principals::default()).await {
        Ok(v) => {
            let e = spectra_core::event::NewEvent::system(
                "ledger.verified",
                serde_json::json!({ "events": v.events, "checkpoints": v.checkpoints, "findings": v.findings.len(), "head": hex::encode(v.head), "first": v.findings.first().map(|f| format!("#{}: {}", f.seq, f.detail)) }),
            );
            let _ = ledger.append_one(e).await;
            Redirect::to("/log").into_response()
        }
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")).into_response(),
    }
}

// ---- Signing in ----

/// `POST /radar/link`, signed by the owner's CLI: a one-time sign-in link.
async fn link(State(st): State<RadarState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, body: Bytes) -> Response {
    let caller = match spectra_mcp::http::authenticate(&st.app, peer, &headers, "POST /radar/link", &body).await {
        Ok(c) => c,
        Err(r) => return r,
    };
    if !matches!(caller.principal, Principal::Owner { .. }) {
        return (StatusCode::FORBIDDEN, axum::Json(serde_json::json!({ "title": "only the owner signs in to Radar" }))).into_response();
    }
    let (Some(key), Some(proof)) = (caller.auth.key, caller.auth.proof) else {
        return StatusCode::UNAUTHORIZED.into_response();
    };
    let code = st.sessions().link(key, proof, now_ms());
    let base = st.app.origins.first().cloned().unwrap_or_default();
    axum::Json(serde_json::json!({ "url": format!("{base}/signin?code={code}"), "valid_ms": session::LINK_MS })).into_response()
}

#[derive(Deserialize)]
struct SigninQuery {
    code: Option<String>,
}

async fn signin(State(st): State<RadarState>, Query(q): Query<SigninQuery>) -> Response {
    let Some(id) = q.code.and_then(|c| st.sessions().open(&c, now_ms())) else {
        return signin_page("That sign-in link has been used or has expired.", StatusCode::UNAUTHORIZED);
    };
    let cookie = format!("{COOKIE}={id}; HttpOnly; SameSite=Strict; Path=/; Max-Age={}", session::SESSION_MS / 1000);
    let mut r = Redirect::to("/").into_response();
    r.headers_mut().insert(header::SET_COOKIE, HeaderValue::from_str(&cookie).unwrap());
    r
}

async fn signout(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response {
    if let Ok(o) = owner(&st, &headers, "live").await {
        if check_form(&st, &o, &headers, &f._csrf).is_none() {
            if let Some(id) = cookie(&headers) {
                st.sessions().close(&id);
            }
        }
    }
    signin_page("Signed out.", StatusCode::OK)
}

// ---- The event stream ----

/// `GET /radar/events`: one SSE message per appended event, carrying its
/// sequence number, kind, actor and subject, never its data (spec §17.3).
async fn events(State(st): State<RadarState>, headers: HeaderMap) -> Response {
    if let Err(r) = owner(&st, &headers, "live").await {
        return r;
    }
    let rx = st.app.daemon.ledger.subscribe();
    let stream = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(|e| async move {
        let e = e.ok()?;
        let data = serde_json::json!({ "seq": e.seq, "kind": e.kind, "actor": e.actor, "subject": e.subject });
        Some(Ok::<_, std::convert::Infallible>(SseEvent::default().data(data.to_string())))
    });
    Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))).into_response()
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/radar/src/lib.rs 83.1%

Keyboard
