Projects with relations from .oxymodules, presence and sessions; projections skip malformed payloads (#5) 2fc88a7b3a5c

-- Projects and how they depend on one another (spec §23.1), and agents'

-- presence and sessions (spec §8.1, §8.2).

CREATE TABLE projects (

    name     TEXT PRIMARY KEY,

    path     TEXT NOT NULL,

    repo_id  TEXT,

    remote   TEXT,

    modules  TEXT NOT NULL,               -- JSON array of the URLs in .oxymodules

    seq      INTEGER NOT NULL

);

CREATE TABLE project_relations (

    project     TEXT NOT NULL,

    depends_on  TEXT NOT NULL,

    PRIMARY KEY (project, depends_on)

);

CREATE TABLE presence (

    agent        TEXT PRIMARY KEY,

    state        TEXT NOT NULL,

    since_ms     INTEGER NOT NULL,

    status_state TEXT,

    status_text  TEXT,

    status_task  TEXT,

    status_ms    INTEGER

);

CREATE TABLE sessions (

    id          TEXT PRIMARY KEY,

    agent       TEXT NOT NULL,

    started_ms  INTEGER NOT NULL,

    ended_ms    INTEGER

);

CREATE INDEX sessions_agent ON sessions(agent);

//! Presence and sessions (spec §8.1, §8.2), kept in memory by the daemon

//! from what agents do, and recorded as events when they change, so the

//! ledger says when an agent went quiet.

use crate::event::NewEvent;

use std::collections::HashMap;

/// A request within this long makes an agent online.

pub const ONLINE_MS: u64 = 60_000;

/// Silence this long ends a session and makes an agent offline.

pub const IDLE_AFTER_MS: u64 = 10 * 60_000;

#[derive(Clone, Debug, Default)]

struct Activity {

    last_request_ms: Option<u64>,

    waiting: u32,

    shims: u32,

    status_ms: Option<u64>,

    blocked: bool,

    paused: bool,

    session: Option<String>,

    state: Option<&'static str>,

}

/// Every agent's activity, and the state last recorded for each.

#[derive(Debug, Default)]

pub struct Presence {

    agents: HashMap<String, Activity>,

}

impl Presence {

    /// A request from `agent`: returns its session id, and a

    /// `agent.session_started` event when this begins a new one.

    pub fn request(&mut self, agent: &str, now: u64, new_session: bool) -> (String, Option<NewEvent>) {

        let a = self.agents.entry(agent.to_string()).or_default();

        let lapsed = a.last_request_ms.is_none_or(|t| now.saturating_sub(t) >= IDLE_AFTER_MS);

        a.last_request_ms = Some(now);

        if a.session.is_some() && !lapsed && !new_session {

            return (a.session.clone().unwrap_or_default(), None);

        }

        let id = crate::ids::ulid();

        a.session = Some(id.clone());

        let e = NewEvent::system("agent.session_started", serde_json::json!({ "agent": agent, "session": id }))

            .subject(agent)

            .session(id.clone());

        (id, Some(e))

    }

    pub fn wait_started(&mut self, agent: &str) {

        self.agents.entry(agent.to_string()).or_default().waiting += 1;

    }

    pub fn wait_ended(&mut self, agent: &str) {

        let a = self.agents.entry(agent.to_string()).or_default();

        a.waiting = a.waiting.saturating_sub(1);

    }

    pub fn shim_connected(&mut self, agent: &str) {

        self.agents.entry(agent.to_string()).or_default().shims += 1;

    }

    pub fn shim_gone(&mut self, agent: &str) {

        let a = self.agents.entry(agent.to_string()).or_default();

        a.shims = a.shims.saturating_sub(1);

    }

    /// The agent set its status (the `status` tool); `None` clears it.

    pub fn status(&mut self, agent: &str, at: Option<u64>) {

        self.agents.entry(agent.to_string()).or_default().status_ms = at;

    }

    pub fn blocked(&mut self, agent: &str, blocked: bool) {

        self.agents.entry(agent.to_string()).or_default().blocked = blocked;

    }

    pub fn paused(&mut self, agent: &str, paused: bool) {

        self.agents.entry(agent.to_string()).or_default().paused = paused;

    }

    /// The state `agent` is in at `now`.

    pub fn state(&self, agent: &str, now: u64) -> &'static str {

        self.agents.get(agent).map(|a| compute(a, now)).unwrap_or("offline")

    }

    /// Events for every change since the last call: `agent.presence` for a

    /// new state, `agent.session_ended` for a session gone quiet.

    pub fn tick(&mut self, now: u64) -> Vec<NewEvent> {

        let mut out = Vec::new();

        let mut names: Vec<&String> = self.agents.keys().collect();

        names.sort();

        let names: Vec<String> = names.into_iter().cloned().collect();

        for name in names {

            let a = self.agents.get_mut(&name).expect("listed");

            let s = compute(a, now);

            if a.state != Some(s) {

                let from = a.state.unwrap_or("offline");

                a.state = Some(s);

                if from != s {

                    out.push(

                        NewEvent::system("agent.presence", serde_json::json!({ "agent": name, "state": s, "from": from }))

                            .subject(name.clone()),

                    );

                }

            }

            let quiet = a.last_request_ms.is_none_or(|t| now.saturating_sub(t) >= IDLE_AFTER_MS);

            if quiet && a.waiting == 0 {

                if let Some(id) = a.session.take() {

                    out.push(

                        NewEvent::system("agent.session_ended", serde_json::json!({ "agent": name, "session": id }))

                            .subject(name.clone())

                            .session(id),

                    );

                }

            }

        }

        out

    }

}

/// Paused, then blocked, then busy, then online, then idle, then offline.

fn compute(a: &Activity, now: u64) -> &'static str {

    let within = |t: Option<u64>, ms: u64| t.is_some_and(|t| now.saturating_sub(t) < ms);

    if a.paused {

        "paused"

    } else if a.blocked {

        "blocked"

    } else if within(a.status_ms, IDLE_AFTER_MS) {

        "busy"

    } else if a.waiting > 0 || within(a.last_request_ms, ONLINE_MS) {

        "online"

    } else if a.shims > 0 || within(a.last_request_ms, IDLE_AFTER_MS) {

        "idle"

    } else {

        "offline"

    }

}

/// Apply presence, status and session events to their tables.

pub async fn project(conn: &mut sqlx::SqliteConnection, e: &crate::event::Event) -> sqlx::Result<()> {

    let p = e.json();

    let s = |k: &str| p.get(k).and_then(|v| v.as_str()).map(String::from);

    // A projection skips what it cannot use: a malformed payload must never

    // stop the ledger from taking events.

    match e.kind.as_str() {

        "agent.presence" if s("agent").is_some() && s("state").is_some() => {

            sqlx::query("INSERT INTO presence (agent, state, since_ms) VALUES (?, ?, ?) ON CONFLICT(agent) DO UPDATE SET state = excluded.state, since_ms = excluded.since_ms")

                .bind(s("agent"))

                .bind(s("state"))

                .bind(e.at_ms as i64)

                .execute(&mut *conn)

                .await?;

        }

        "agent.status" if e.actor.starts_with("agent:") => {

            let agent = e.actor.strip_prefix("agent:").map(String::from);

            sqlx::query(

                "INSERT INTO presence (agent, state, since_ms, status_state, status_text, status_task, status_ms) VALUES (?, 'online', ?, ?, ?, ?, ?) \

                 ON CONFLICT(agent) DO UPDATE SET status_state = excluded.status_state, status_text = excluded.status_text, status_task = excluded.status_task, status_ms = excluded.status_ms",

            )

            .bind(agent)

            .bind(e.at_ms as i64)

            .bind(s("state"))

            .bind(s("text"))

            .bind(s("task"))

            .bind(e.at_ms as i64)

            .execute(&mut *conn)

            .await?;

        }

        "agent.session_started" if s("agent").is_some() && s("session").is_some() => {

            sqlx::query("INSERT OR IGNORE INTO sessions (id, agent, started_ms) VALUES (?, ?, ?)")

                .bind(s("session"))

                .bind(s("agent"))

                .bind(e.at_ms as i64)

                .execute(&mut *conn)

                .await?;

        }

        "agent.session_ended" => {

            sqlx::query("UPDATE sessions SET ended_ms = ? WHERE id = ?")

                .bind(e.at_ms as i64)

                .bind(s("session"))

                .execute(&mut *conn)

                .await?;

        }

        _ => {}

    }

    Ok(())

}

#[cfg(test)]

mod tests {

    use super::*;

    fn kinds(v: &[NewEvent]) -> Vec<String> {

        v.iter().map(|e| format!("{} {}", e.kind, e.payload.get("state").and_then(|s| s.as_str()).unwrap_or(""))).collect()

    }

    #[test]

    fn presence_moves_through_its_states_and_records_each_change() {

        let mut p = Presence::default();

        let t = 1_000_000;

        let (s1, started) = p.request("claude", t, false);

        assert!(started.is_some());

        assert_eq!(kinds(&p.tick(t)), ["agent.presence online"]);

        assert!(p.tick(t + 1).is_empty(), "no change, no event");

        assert_eq!(p.state("claude", t + ONLINE_MS + 1), "idle");

        assert_eq!(kinds(&p.tick(t + ONLINE_MS + 1)), ["agent.presence idle"]);

        p.status("claude", Some(t + ONLINE_MS + 2));

        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "busy");

        p.blocked("claude", true);

        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "blocked");

        p.blocked("claude", false);

        p.status("claude", None);

        p.paused("claude", true);

        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "paused");

        p.paused("claude", false);

        let late = t + IDLE_AFTER_MS + 5;

        let ev = p.tick(late);

        assert_eq!(kinds(&ev), ["agent.presence offline", "agent.session_ended "]);

        let (s2, started) = p.request("claude", late + 1, false);

        assert!(started.is_some() && s2 != s1, "a new session after ten minutes' silence");

        let (s3, none) = p.request("claude", late + 2, false);

        assert!(none.is_none() && s3 == s2);

        let (s4, again) = p.request("claude", late + 3, true);

        assert!(again.is_some() && s4 != s3, "brief(new_session: true)");

    }

    #[test]

    fn a_waiting_or_connected_agent_is_not_offline() {

        let mut p = Presence::default();

        p.request("kimi", 0, false);

        p.wait_started("kimi");

        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "online");

        p.wait_ended("kimi");

        p.shim_connected("kimi");

        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "idle");

        p.shim_gone("kimi");

        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "offline");

    }

}

//! Projects (spec §23.1): checkouts the agents work in, with the

//! repository id and remote read from `.oxy/`, and dependencies read from

//! `.oxymodules`, matched to other registered projects by remote URL.

use crate::event::Event;

use anyhow::{Context, Result};

use serde_json::Value as Json;

use sqlx::SqliteConnection;

use std::path::Path;

/// What a checkout says about itself.

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

pub struct Checkout {

    pub repo_id: Option<String>,

    pub remote: Option<String>,

    /// The URLs of the subrepos it pins (`.oxymodules`).

    pub modules: Vec<String>,

}

/// Read `.oxy/id`, the remote (`origin`, else the first) and `.oxymodules`.

pub fn inspect(path: &Path) -> Result<Checkout> {

    if !path.is_dir() {

        anyhow::bail!("{} is not a directory", path.display());

    }

    let oxy = path.join(".oxy");

    let repo_id = std::fs::read_to_string(oxy.join("id")).ok().map(|s| s.trim().to_string()).filter(|s| !s.is_empty());

    let mut remotes: Vec<(String, String)> = Vec::new();

    if let Ok(text) = std::fs::read_to_string(oxy.join("config")) {

        for line in text.lines() {

            if let Some((k, v)) = line.split_once('=') {

                let k = k.trim();

                if let Some(name) = k.strip_prefix("remote.").and_then(|r| r.strip_suffix(".url")) {

                    remotes.push((name.to_string(), v.trim().trim_matches('"').to_string()));

                }

            }

        }

    }

    let remote = remotes.iter().find(|(n, _)| n == "origin").or(remotes.first()).map(|(_, u)| u.clone());

    let modules = std::fs::read_to_string(path.join(".oxymodules"))

        .map(|t| {

            t.lines()

                .filter_map(|l| l.trim().strip_prefix("url").map(|r| r.trim_start().trim_start_matches('=').trim().to_string()))

                .collect()

        })

        .unwrap_or_default();

    Ok(Checkout { repo_id, remote, modules })

}

/// A project's name: the directory's, lower-cased, made valid.

pub fn default_name(path: &Path) -> Result<String> {

    let base = path.canonicalize()?.file_name().and_then(|n| n.to_str()).map(str::to_lowercase).context("the path has no name")?;

    Ok(base.chars().map(|c| if c.is_ascii_alphanumeric() || c == '-' { c } else { '-' }).collect())

}

fn same_repo(a: &str, b: &str) -> bool {

    let n = |s: &str| s.trim_end_matches('/').trim_end_matches(".git").to_lowercase();

    n(a) == n(b)

}

/// Apply `project.added` to the tables, deriving relations both ways.

pub async fn project(conn: &mut SqliteConnection, e: &Event) -> sqlx::Result<()> {

    if e.kind != "project.added" {

        return Ok(());

    }

    let p = e.json();

    let s = |k: &str| p.get(k).and_then(Json::as_str).map(String::from);

    let Some(name) = s("name") else { return Ok(()) };

    let modules: Vec<String> = p

        .get("modules")

        .and_then(Json::as_array)

        .map(|a| a.iter().filter_map(|m| m.as_str().map(String::from)).collect())

        .unwrap_or_default();

    let remote = s("remote");

    sqlx::query("INSERT OR REPLACE INTO projects (name, path, repo_id, remote, modules, seq) VALUES (?, ?, ?, ?, ?, ?)")

        .bind(&name)

        .bind(s("path").unwrap_or_default())

        .bind(s("repo_id"))

        .bind(&remote)

        .bind(serde_json::to_string(&modules).unwrap_or_default())

        .bind(e.seq as i64)

        .execute(&mut *conn)

        .await?;

    let others: Vec<(String, Option<String>, String)> =

        sqlx::query_as("SELECT name, remote, modules FROM projects WHERE name != ?").bind(&name).fetch_all(&mut *conn).await?;

    for (other, other_remote, other_modules) in others {

        if other_remote.as_deref().is_some_and(|r| modules.iter().any(|m| same_repo(m, r))) {

            sqlx::query("INSERT OR IGNORE INTO project_relations (project, depends_on) VALUES (?, ?)")

                .bind(&name)

                .bind(&other)

                .execute(&mut *conn)

                .await?;

        }

        let theirs: Vec<String> = serde_json::from_str(&other_modules).unwrap_or_default();

        if remote.as_deref().is_some_and(|r| theirs.iter().any(|m| same_repo(m, r))) {

            sqlx::query("INSERT OR IGNORE INTO project_relations (project, depends_on) VALUES (?, ?)")

                .bind(&other)

                .bind(&name)

                .execute(&mut *conn)

                .await?;

        }

    }

    Ok(())

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn a_checkout_reports_its_id_remote_and_subrepos() {

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

        std::fs::create_dir_all(dir.path().join(".oxy")).unwrap();

        std::fs::write(dir.path().join(".oxy/id"), "97e6").unwrap();

        std::fs::write(

            dir.path().join(".oxy/config"),

            "remote.earth.url = \"https://hub/a/hub\"\nremote.origin.url = \"https://hub/a/x\"\n",

        )

        .unwrap();

        std::fs::write(dir.path().join(".oxymodules"), "[deps/flow]\nurl = https://hub/a/flow\nbookmark = main\n").unwrap();

        let c = inspect(dir.path()).unwrap();

        assert_eq!(

            c,

            Checkout { repo_id: Some("97e6".into()), remote: Some("https://hub/a/x".into()), modules: vec!["https://hub/a/flow".into()] }

        );

        assert!(same_repo("https://hub/a/flow/", "https://HUB/a/flow.git"));

    }

}

//! `spectra project`: the checkouts agents work in (spec §23.1).

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

use super::{ui, Ctx};

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

use spectra_core::keys::KeyPair;

use spectra_core::ledger::Ledger;

use spectra_core::principals::signed_event;

use spectra_core::projects;

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

pub async fn add(ctx: &Ctx, path: &Path, name: Option<String>, remote: Option<String>, owner_home_flag: Option<PathBuf>) -> Result<i32> {

    let path = path.canonicalize().with_context(|| format!("{} does not exist", path.display()))?;

    let name = match name {

        Some(n) => n,

        None => projects::default_name(&path)?,

    };

    if !super::agent::valid_name(&name) {

        bail!("a project's name is lower-case letters, digits and dashes (--name)");

    }

    let checkout = projects::inspect(&path)?;

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

    let exists: Option<String> = sqlx::query_scalar("SELECT name FROM projects WHERE name = ? OR path = ?")

        .bind(&name)

        .bind(path.display().to_string())

        .fetch_optional(ledger.pool())

        .await?;

    if let Some(e) = exists {

        bail!("{} is already a project ({e})", path.display());

    }

    let home = owner_home(ctx, owner_home_flag);

    refresh_owner(&ledger, &home).await?;

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

    let remote = remote.or(checkout.remote.clone());

    let payload = serde_json::json!({

        "name": name, "path": path.display().to_string(), "repo_id": checkout.repo_id, "remote": remote, "modules": checkout.modules,

    });

    ledger.append_one(signed_event(&owner, "human", "project.added", payload).project(name.clone()).subject(name.clone())).await?;

    ledger.checkpoint().await?;

    println!("{} {}  {}", ui::done("Added"), ui::subject(&name), ui::path(path.display()));

    if let Some(r) = &remote {

        println!("{}", ui::kv("remote", r));

    }

    if let Some(id) = &checkout.repo_id {

        println!("{}", ui::kv("repo", ui::hash(id)));

    }

    let (deps, users) = relations(&ledger, &name).await?;

    if !deps.is_empty() {

        println!("{}", ui::kv("pins", deps.join(", ")));

    }

    if !users.is_empty() {

        println!("{}", ui::kv("pinned by", users.join(", ")));

    }

    if checkout.repo_id.is_none() {

        println!("  {} {}", ui::note(), ui::dim("not an Oxygen checkout: relations come from .oxymodules, so it has none"));

    }

    Ok(0)

}

async fn relations(ledger: &Ledger, name: &str) -> Result<(Vec<String>, Vec<String>)> {

    let deps = sqlx::query_scalar("SELECT depends_on FROM project_relations WHERE project = ? ORDER BY 1")

        .bind(name)

        .fetch_all(ledger.pool())

        .await?;

    let users = sqlx::query_scalar("SELECT project FROM project_relations WHERE depends_on = ? ORDER BY 1")

        .bind(name)

        .fetch_all(ledger.pool())

        .await?;

    Ok((deps, users))

}

pub async fn list(ctx: &Ctx, json: bool) -> Result<i32> {

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

    let rows: Vec<(String, String, Option<String>)> =

        sqlx::query_as("SELECT name, path, remote FROM projects ORDER BY name").fetch_all(ledger.pool()).await?;

    let mut out = Vec::new();

    for (name, path, remote) in rows {

        let (deps, users) = relations(&ledger, &name).await?;

        let agents: Vec<String> = sqlx::query_scalar("SELECT name FROM agents WHERE project = ? AND disabled_ms IS NULL ORDER BY name")

            .bind(&name)

            .fetch_all(ledger.pool())

            .await?;

        out.push(serde_json::json!({ "name": name, "path": path, "remote": remote, "depends_on": deps, "depended_on_by": users, "agents": agents }));

    }

    if json {

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

        return Ok(0);

    }

    if out.is_empty() {

        println!("{} {}", ui::note(), ui::dim("no projects yet: spectra project add PATH"));

    }

    for p in out {

        let list =

            |k: &str| p[k].as_array().map(|a| a.iter().filter_map(|v| v.as_str()).collect::<Vec<_>>().join(", ")).unwrap_or_default();

        println!(

            "{} {}  {}",

            ui::glyph("▸", ">"),

            ui::bookmark(p["name"].as_str().unwrap_or("")),

            ui::path(p["path"].as_str().unwrap_or(""))

        );

        for (k, label) in [("depends_on", "pins"), ("depended_on_by", "pinned by"), ("agents", "agents")] {

            let v = list(k);

            if !v.is_empty() {

                println!("{}", ui::kv(label, v));

            }

        }

    }

    Ok(0)

}

//! Projects from the command line (spec §23.1): relations come from

//! `.oxymodules`, matched by remote, whichever is registered first; an agent

//! needs a registered project.

use spectra_core::keys::KeyPair;

use std::path::Path;

use std::process::{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))

}

/// A checkout as `oxy` leaves one: an id, a remote, and the subrepos it pins.

fn checkout(at: &Path, remote: &str, pins: &[&str]) {

    std::fs::create_dir_all(at.join(".oxy")).unwrap();

    std::fs::write(at.join(".oxy/id"), "0123456789abcdef0123456789abcdef").unwrap();

    std::fs::write(at.join(".oxy/config"), format!("remote.origin.url = \"{remote}\"\n")).unwrap();

    let modules: String = pins.iter().enumerate().map(|(i, u)| format!("[deps/{i}]\nurl = {u}\nbookmark = main\n")).collect();

    std::fs::write(at.join(".oxymodules"), modules).unwrap();

}

fn setup() -> tempfile::TempDir {

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

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

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

    dir

}

fn projects(d: &Path) -> serde_json::Value {

    serde_json::from_slice(&run(d, &["project", "list", "--json"]).stdout).unwrap()

}

#[test]

fn relations_are_found_whichever_project_comes_first() {

    for hub_first in [false, true] {

        let dir = setup();

        let d = dir.path();

        checkout(&d.join("oxygen"), "https://hub.example/andy/oxygen", &[]);

        checkout(&d.join("oxygenhub"), "https://hub.example/andy/oxygenhub", &["https://hub.example/andy/oxygen"]);

        let order = if hub_first { ["oxygenhub", "oxygen"] } else { ["oxygen", "oxygenhub"] };

        for p in order {

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

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

        }

        let list = projects(d);

        let hub = list.as_array().unwrap().iter().find(|p| p["name"] == "oxygenhub").unwrap();

        let oxy = list.as_array().unwrap().iter().find(|p| p["name"] == "oxygen").unwrap();

        assert_eq!(hub["depends_on"], serde_json::json!(["oxygen"]), "hub first: {hub_first}");

        assert_eq!(oxy["depended_on_by"], serde_json::json!(["oxygenhub"]));

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

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

    }

}

#[test]

fn a_project_is_registered_once_and_an_agent_needs_one() {

    let dir = setup();

    let d = dir.path();

    checkout(&d.join("oxygen"), "https://hub.example/andy/oxygen", &[]);

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

    let again = run(d, &["project", "add", d.join("oxygen").to_str().unwrap(), "--name", "other"]);

    assert!(!again.status.success() && text(&again).contains("already a project"), "{}", text(&again));

    let o = run(

        d,

        &["agent", "add", "kimi", "--kind", "kimi-code", "--project", "nowhere", "--home", d.join("kimi").to_str().unwrap(), "--wait", "0"],

    );

    assert!(!o.status.success() && text(&o).contains("no project named nowhere"), "{}", text(&o));

    assert!(!d.join("kimi").exists(), "no key is made for an agent that cannot be added");

}

- **Added: projects, presence and sessions (#5).** `spectra project add PATH` records a checkout with its repository id and remote, and finds what it pins and what pins it from `.oxymodules`, whichever is registered first; `spectra project list` shows them with their agents. An agent needs a registered project. Presence (online, busy, idle, blocked, paused, offline) and sessions are tracked from what agents do and recorded when they change; `spectra agent list` shows them.

pub mod presence;

pub mod projects;

    // Who is who is only ever said by a signature; `verify` checks which.

    if e.auth.method != Method::Signature {

        return Ok(());

    }

            if p.get("name").and_then(Json::as_str).is_none_or(str::is_empty) {

                return Ok(());

            }

pub const TABLES: &[&str] = &["event_counts", "identities", "agents", "projects", "project_relations", "presence", "sessions"];

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

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

    // Kinds that projections read, with payloads that are often malformed:

    // a projection must skip what it cannot use, never refuse the append.

    let kinds = [

        "message.posted",

        "agent.presence",

        "agent.status",

        "agent.session_started",

        "agent.session_ended",

        "agent.added",

        "agent.disabled",

        "identity.owner",

        "identity.rotated",

        "project.added",

        "task.created",

        "tool.called",

    ];

#[tokio::test]

async fn an_unsigned_claim_to_be_the_owner_or_an_agent_never_reaches_the_tables() {

    let w = world().await;

    let fake = KeyPair::generate("mallory", "");

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

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

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

    w.ledger.append_one(NewEvent::system("agent.added", p)).await.unwrap();

    w.ledger.append_one(NewEvent::system("identity.rotated", identity_payload(&fake.identity))).await.unwrap();

    let agents: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM agents").fetch_one(w.ledger.pool()).await.unwrap();

    let owners: Vec<String> =

        sqlx::query_scalar("SELECT name FROM identities WHERE role = 'owner' AND current = 1").fetch_all(w.ledger.pool()).await.unwrap();

    assert_eq!(agents, 0);

    assert_eq!(owners, ["Andy McCall"]);

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

    assert_eq!(v.findings.len(), 2, "both are reported: {:?}", v.findings);

}

- [x] #5 Projects, presence and sessions — done 2026-11-02: project add/list with relations from .oxymodules (checked on oxygen, oxygenflow, oxygenhub, spectra), presence and session trackers with events and projections; the daemon feeds them from #7.

- 2026-11-02 Andy asked for an oxy-style CLI: every command now reports through oxy's ui module. #5 done.

        let known: Option<String> =

            sqlx::query_scalar("SELECT name FROM projects WHERE name = ?").bind(&a.project).fetch_optional(ledger.pool()).await?;

        if known.is_none() {

            bail!("no project named {} (spectra project add PATH --name {})", a.project, a.project);

        }

}

/// An agent as `agent list` reads it, with its presence.

#[derive(sqlx::FromRow)]

struct AgentRow {

    name: String,

    key: String,

    kind: String,

    project: String,

    disabled_ms: Option<i64>,

    presence: Option<String>,

    status: Option<String>,

    let rows: Vec<AgentRow> = sqlx::query_as(

        "SELECT a.name, a.key, a.kind, a.project, a.disabled_ms, p.state AS presence, p.status_text AS status FROM agents a LEFT JOIN presence p ON p.agent = a.name ORDER BY a.name",

    )

    .fetch_all(ledger.pool())

    .await?;

    for AgentRow { name, key, kind, project, disabled_ms: disabled, presence, status } in rows {

        let presence = if disabled.is_some() { "offline".to_string() } else { presence.unwrap_or_else(|| "offline".into()) };

        out.push(serde_json::json!({ "name": name, "display": who, "key": key, "kind": kind, "project": project, "state": state, "presence": presence, "status": status, "delegation_expires": expires, "disabled": disabled.is_some() }));

        println!(

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

            marks::health(state),

            marks::presence(a["presence"].as_str().unwrap_or("offline")),

            ui::subject(name),

            ui::sep(),

            ui::author(owner)

        );

        if let Some(st) = a["status"].as_str() {

            println!("  {} {}", ui::note(), st);

        }

pub mod project;

    /// The checkouts agents work in: add, list

    Project {

        #[command(subcommand)]

        command: ProjectCmd,

    },

pub enum ProjectCmd {

    /// Register a checkout: its repository id, remote and pinned subrepos

    Add {

        path: PathBuf,

        /// Its name (default: the directory's)

        #[arg(long)]

        name: Option<String>,

        /// Its remote URL (default: the checkout's origin)

        #[arg(long)]

        remote: Option<String>,

        #[arg(long)]

        owner_home: Option<PathBuf>,

    },

    /// Every project with what it pins, what pins it, and its agents

    List {

        #[arg(long)]

        json: bool,

    },

}

#[derive(Subcommand)]

            }

        }),

        Command::Project { command } => runtime()?.block_on(async {

            match command {

                ProjectCmd::Add { path, name, remote, owner_home } => project::add(&ctx, &path, name, remote, owner_home).await,

                ProjectCmd::List { json } => project::list(&ctx, json).await,

    let checkout = d.join("oxygen");

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

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

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