diff --git a/AGENTS.md b/AGENTS.md index 3182f10..be16c02 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ Studio's job, not Huginn's. derived status included, over `cultcache-rs` and `eureka-pipeline`. `huginn-daemon` serves one mind over CultNet RUDP through `cultnet-rs`, and reaches the leaf and the store type through `huginn-mind` so the workspace's one CultLib revision (root `[workspace.dependencies]`) decides both. `eureka-state` is the client core: one call - to a daemon over CultNet, holding no state. + to a daemon over CultNet, holding no state; its `huginn` binary is the CLI agents reach the mind through, and it holds no state either. - Owned: an instance's memory documents, their admission and the read side that derives their status, in `huginn-mind`, over a redb CultCache store at `/minds//mind.redb`; diff --git a/Cargo.lock b/Cargo.lock index 7731775..c219ca6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -86,6 +86,56 @@ dependencies = [ "libc", ] +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + [[package]] name = "anyhow" version = "1.0.104" @@ -257,6 +307,52 @@ dependencies = [ "inout", ] +[[package]] +name = "clap" +version = "4.6.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "473c7e07f409a8d772161724aa8db6a765a2532a70f9667eeb7b49d3d02fbdca" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b48fea5a88e9ae728a2dcbedbfc0e730f7d60da42e1cb049a83c9fb8b789889" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d012d2b9d65aca7f18f4d9878a045bc17899bba951561ba5ec3c2ba1eed9a061" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + [[package]] name = "const-oid" version = "0.9.6" @@ -514,6 +610,7 @@ version = "0.1.0" dependencies = [ "anyhow", "chrono", + "clap", "cultnet-rs", "huginn-daemon", "huginn-mind", @@ -883,6 +980,12 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "itoa" version = "1.0.18" @@ -1105,6 +1208,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "opaque-debug" version = "0.3.1" @@ -1968,6 +2077,12 @@ version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e" +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + [[package]] name = "uuid" version = "1.26.1" diff --git a/README.md b/README.md index 41a6815..028f0b2 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,9 @@ A Cargo workspace of three crates: the process, and no rule. The operation envelope both sides speak is `huginn-mind`'s. - `crates/eureka-state`: the client core, `HuginnClient::call`: one request to a - daemon over CultNet RUDP, a deferred answer resolved, and no state. + daemon over CultNet RUDP, a deferred answer resolved, and no state. Its + `huginn` binary is the agents' CLI: whoami, view, query and named recipes, + admit with read-back, schema. `huginn-mind` is live: it opens one instance's store (an owned redb CultCache at `/minds//mind.redb`, locked for the mind's lifetime), diff --git a/crates/eureka-state/Cargo.toml b/crates/eureka-state/Cargo.toml index a1c5fb6..74f1143 100644 --- a/crates/eureka-state/Cargo.toml +++ b/crates/eureka-state/Cargo.toml @@ -1,32 +1,50 @@ -[package] -name = "eureka-state" -version = "0.1.0" -edition.workspace = true -license.workspace = true -publish.workspace = true - -# The client core (`lib.rs`): one call over CultNet to a Huginn daemon. It owns -# the transport and nothing of the mind. The operation envelope, the wire -# vocabulary and the deferred-body bound are `huginn-mind`'s, so the client and -# the daemon cannot spell them differently; the content fetch and its -# verification are `cultnet-rs`'s. -# -# The MCP server (`main.rs`, the `eureka-state` binary) is the client's one -# consumer: six tools over stdio, one per wire operation, holding no state. -[dependencies] -anyhow = "1" -cultnet-rs.workspace = true -huginn-mind = { path = "../huginn-mind" } -rmcp = { version = "2.2.0", default-features = false, features = ["server", "macros", "transport-io"] } -rmp-serde = "1" -schemars = "1" -serde = { version = "1", features = ["derive"] } -serde_json = "1" -tokio = { version = "1", features = ["rt", "macros", "io-std"] } - -# Tests run a real daemon in-process against a temporary mind, and the real -# binary against it. -[dev-dependencies] -chrono = "0.4.44" -huginn-daemon = { path = "../huginn-daemon" } -tempfile = "3" +[package] +name = "eureka-state" +version = "0.1.0" +edition.workspace = true +license.workspace = true +publish.workspace = true + +# The client core (`lib.rs`): one call over CultNet to a Huginn daemon. It owns +# the transport and nothing of the mind. The operation envelope, the wire +# vocabulary and the deferred-body bound are `huginn-mind`'s, so the client and +# the daemon cannot spell them differently; the content fetch and its +# verification are `cultnet-rs`'s. +# +# The crate is the client core (`lib.rs`) and its consumers: the `huginn` CLI +# (`src/bin/huginn/`), the shell door every agent uses, and until +# delete-eureka-state-mcp the MCP server (`main.rs`, the `eureka-state` binary). +# Both hold no state. +# `cli` (default) is the `huginn` binary's argument parser; a library consumer +# takes `default-features = false` and compiles none of it. +[features] +default = ["cli"] +cli = ["dep:clap"] + +[[bin]] +name = "huginn" +path = "src/bin/huginn/main.rs" +required-features = ["cli"] + +[[test]] +name = "cli" +required-features = ["cli"] + +[dependencies] +anyhow = "1" +clap = { version = "4", features = ["derive"], optional = true } +cultnet-rs.workspace = true +huginn-mind = { path = "../huginn-mind" } +rmcp = { version = "2.2.0", default-features = false, features = ["server", "macros", "transport-io"] } +rmp-serde = "1" +schemars = "1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["rt", "macros", "io-std"] } + +# Tests run a real daemon in-process against a temporary mind, and the real +# binary against it. +[dev-dependencies] +chrono = "0.4.44" +huginn-daemon = { path = "../huginn-daemon" } +tempfile = "3" diff --git a/crates/eureka-state/src/bin/huginn/config.rs b/crates/eureka-state/src/bin/huginn/config.rs new file mode 100644 index 0000000..d20e0e9 --- /dev/null +++ b/crates/eureka-state/src/bin/huginn/config.rs @@ -0,0 +1,67 @@ +//! The CLI's configuration: the mind is the session's environment, never an +//! argument. `EUREKA_INSTANCE` names the instance and `HUGINN_ENDPOINT` +//! (`rudp://:`) the daemon that holds it. + +use std::net::SocketAddr; +use std::time::Duration; + +use eureka_state::HuginnClient; +use huginn_mind::eureka_pipeline::Slug; +use huginn_mind::{HuginnMindRequest, HuginnMindResponse}; + +use crate::trouble::Trouble; + +/// The `provenance.tool` every admission from this binary carries. +pub const TOOL: &str = "huginn"; +/// How long one call may take, connection to last byte. +const TIMEOUT: Duration = Duration::from_secs(15); + +/// The two environment values as given, and the client they make when both +/// are valid. +pub struct Settings { + pub instance: Option, + pub endpoint: Option, + client: Result, +} + +impl Settings { + pub fn from_env() -> Self { + let instance = variable("EUREKA_INSTANCE"); + let endpoint = variable("HUGINN_ENDPOINT"); + let shown = |value: &Result, String>| value.as_ref().ok().cloned().flatten(); + let (shown_instance, shown_endpoint) = (shown(&instance), shown(&endpoint)); + Self { instance: shown_instance, endpoint: shown_endpoint, client: client(instance, endpoint) } + } + + pub fn client(&self) -> Result<&HuginnClient, Trouble> { + self.client.as_ref().map_err(|detail| Trouble::Misconfigured(detail.clone())) + } + + /// One wire call for the configured instance. The request is built from + /// that instance, never from anything the caller sent. No retry. + pub fn ask(&self, request: impl FnOnce(&Slug) -> HuginnMindRequest) -> Result { + let client = self.client()?; + client.call(request(client.instance())).map_err(Trouble::from) + } +} + +/// An environment variable: its value, `None` when unset, or why it cannot be read. +fn variable(name: &str) -> Result, String> { + match std::env::var(name) { + Ok(value) => Ok(Some(value)), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(std::env::VarError::NotUnicode(_)) => Err(format!("{name} is not valid UTF-8")), + } +} + +fn client(instance: Result, String>, endpoint: Result, String>) -> Result { + let instance = instance?.ok_or("EUREKA_INSTANCE is not set")?; + let slug = Slug(instance); + slug.validate_slug().map_err(|refusal| format!("EUREKA_INSTANCE is not a valid instance name: {refusal:?}"))?; + let endpoint = endpoint?.ok_or("HUGINN_ENDPOINT is not set")?; + let address = endpoint + .strip_prefix("rudp://") + .and_then(|address| address.parse::().ok()) + .ok_or_else(|| format!("HUGINN_ENDPOINT is not rudp://:: {endpoint}"))?; + Ok(HuginnClient::new(address, slug, TIMEOUT)) +} diff --git a/crates/eureka-state/src/bin/huginn/door.rs b/crates/eureka-state/src/bin/huginn/door.rs new file mode 100644 index 0000000..d460da1 --- /dev/null +++ b/crates/eureka-state/src/bin/huginn/door.rs @@ -0,0 +1,204 @@ +//! The text door: agent-written JSON becomes typed values here, strictly, and +//! nowhere else. The store's decode stays lenient (serde ignores unknown +//! keys), so a misspelled key would be dropped and the batch committed +//! without it. The door refuses instead: every key path present in the input +//! with a value and absent from the typed value's re-serialisation is named in +//! the refusal. Paths only, never values. + +use std::io::Read; + +use cultnet_rs::Selection; +use huginn_mind::eureka_pipeline::{PipelineDocument, PipelineKind}; +use serde::Serialize; +use serde::de::DeserializeOwned; +use serde_json::{Value, error::Category}; + +use crate::trouble::Trouble; + +/// The text of an argument that is a literal, `@` or `-` for stdin. +pub fn argument(arg: &str) -> Result { + match arg.strip_prefix('@') { + Some(path) => file(path), + None if arg == "-" => stdin(), + None => Ok(arg.to_string()), + } +} + +/// The text of a file argument, or stdin when it is `-`. +pub fn file_or_stdin(arg: &str) -> Result { + if arg == "-" { stdin() } else { file(arg) } +} + +fn file(path: &str) -> Result { + std::fs::read_to_string(path).map_err(|error| Trouble::invalid(format!("cannot read {path}: {}", error.kind()))) +} + +fn stdin() -> Result { + let mut text = String::new(); + std::io::stdin() + .read_to_string(&mut text) + .map_err(|error| Trouble::invalid(format!("cannot read stdin: {}", error.kind())))?; + Ok(text) +} + +/// A batch: a JSON array of `{kind, value}` documents, or one such object. +pub fn documents(text: &str) -> Result, Trouble> { + let value = parse(text)?; + let (items, typed): (&[Value], Vec) = match &value { + Value::Array(items) => (items.as_slice(), objects(items).and_then(|()| from_text(text, "documents"))?), + Value::Object(_) => (std::slice::from_ref(&value), vec![from_text(text, "documents[0]")?]), + _ => return Err(Trouble::invalid("the documents must be a JSON array of {kind, value} objects")), + }; + let kept = reserialise(&typed, "documents")?; + let mut lost = Vec::new(); + for (index, (given, held)) in items.iter().zip(&kept).enumerate() { + dropped(given, held, &format!("documents[{index}]"), &mut lost); + } + refuse_lost(&lost)?; + Ok(typed) +} + +pub fn selection(text: &str) -> Result { + let value = parse(text)?; + if !value.is_object() { + return Err(Trouble::invalid("selection must be a JSON object")); + } + let typed: Selection = from_text(text, "selection")?; + let kept = reserialise(std::slice::from_ref(&typed), "selection")?; + let mut lost = Vec::new(); + dropped(&value, &kept[0], "selection", &mut lost); + refuse_lost(&lost)?; + Ok(typed) +} + +pub fn selection_value(value: &Value) -> Result { + selection(&value.to_string()) +} + +/// The kind a pipeline id names in its middle segment (`::`). +pub fn kind_of(id: &str) -> Result { + let mut segments = id.splitn(3, ':'); + let (_, Some(name), Some(_)) = (segments.next(), segments.next(), segments.next()) else { + return Err(Trouble::invalid("a pipeline id is ::")); + }; + PipelineKind::ALL + .iter() + .copied() + .find(|kind| kind.name() == name) + .ok_or_else(|| Trouble::invalid("the id's kind segment is not a pipeline kind (`huginn schema` lists them)")) +} + +fn parse(text: &str) -> Result { + serde_json::from_str(text).map_err(|error| Trouble::invalid(format!("not JSON: {}", place(&error)))) +} + +/// Where and what kind of failure a decode hit, never what the input said: +/// serde's own message quotes the offending value whole. +fn place(error: &serde_json::Error) -> String { + let category = match error.classify() { + Category::Io => "unreadable input", + Category::Syntax => "syntax", + Category::Data => "a value of the wrong type, an unknown variant or a missing field", + Category::Eof => "unexpected end", + }; + format!("{category} at line {} column {}", error.line(), error.column()) +} + +/// Serde would also read a struct from an array, positionally, so only an +/// object is a document or a selection. +fn objects(items: &[Value]) -> Result<(), Trouble> { + match items.iter().position(|item| !item.is_object()) { + None => Ok(()), + Some(index) => Err(Trouble::invalid(format!("documents[{index}] must be a JSON object"))), + } +} + +/// The typed value of the text, decoded from the text itself so the failure +/// has a real line and column. +fn from_text(text: &str, at: &str) -> Result { + serde_json::from_str(text).map_err(|error| Trouble::invalid(format!("{at} does not decode: {}", place(&error)))) +} + +/// What the typed values hold, as the JSON items to compare with the input. +fn reserialise(typed: &[T], at: &str) -> Result, Trouble> { + typed + .iter() + .map(|item| serde_json::to_value(item).map_err(|_| Trouble::Internal(format!("{at} does not re-serialise")))) + .collect() +} + +fn refuse_lost(paths: &[String]) -> Result<(), Trouble> { + if paths.is_empty() { + Ok(()) + } else { + Err(Trouble::invalid(format!("keys the schema does not know (`huginn schema` prints it): {}", paths.join(", ")))) + } +} + +/// The paths of `input` that carry a value and are absent from `kept`. +fn dropped(input: &Value, kept: &Value, path: &str, out: &mut Vec) { + match (input, kept) { + (Value::Object(given), Value::Object(held)) => { + for (key, value) in given { + let here = format!("{path}.{key}"); + match held.get(key) { + Some(held) => dropped(value, held, &here, out), + None if !empty(value) => out.push(here), + None => {} + } + } + } + (Value::Array(given), Value::Array(held)) => { + for (index, (value, held)) in given.iter().zip(held).enumerate() { + dropped(value, held, &format!("{path}[{index}]"), out); + } + } + _ => {} + } +} + +/// A value whose loss changes nothing: the spellings of "not given". +fn empty(value: &Value) -> bool { + match value { + Value::Null => true, + Value::String(text) => text.is_empty(), + Value::Array(items) => items.is_empty(), + Value::Object(fields) => fields.is_empty(), + Value::Bool(_) | Value::Number(_) => false, + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + fn paths(input: &Value, kept: &Value) -> Vec { + let mut out = Vec::new(); + dropped(input, kept, "d", &mut out); + out.sort(); + out + } + + #[test] + fn a_dropped_key_with_a_value_is_named_by_path() { + let input = json!({ "a": { "b": [ { "c": 1, "typo": "CANARY" } ] }, "top": 7 }); + let kept = json!({ "a": { "b": [ { "c": 1 } ] } }); + assert_eq!(paths(&input, &kept), vec!["d.a.b[0].typo".to_string(), "d.top".to_string()]); + } + + #[test] + fn a_dropped_key_without_a_value_is_not_a_loss() { + let input = json!({ "n": null, "s": "", "l": [], "o": {}, "zero": 0, "no": false }); + assert_eq!(paths(&input, &json!({})), vec!["d.no".to_string(), "d.zero".to_string()]); + } + + #[test] + fn kinds_come_from_the_middle_segment() { + assert_eq!(kind_of("eureka-substrate:cut_spec:cut-x.r1").ok(), Some(PipelineKind::CutSpec)); + assert!(kind_of("eureka-substrate:nonsense:x").is_err()); + assert!(kind_of("two:segments").is_err()); + assert!(kind_of("campaign").is_err()); + } +} diff --git a/crates/eureka-state/src/bin/huginn/main.rs b/crates/eureka-state/src/bin/huginn/main.rs new file mode 100644 index 0000000..f2bb165 --- /dev/null +++ b/crates/eureka-state/src/bin/huginn/main.rs @@ -0,0 +1,398 @@ +//! `huginn`: the agents' door to Huginn's mind, as shell commands. +//! +//! A caller of `HuginnClient` and nothing else. It owns argument parsing, the +//! strict decode of agent-written JSON into the leaf's types (`door`), the +//! recipe selections (`recipes`), the text rendering (`render`), the exit code +//! and the read-back after an admission. It owns no state, writes no file and +//! reads no configuration but two environment variables: `EUREKA_INSTANCE` +//! names the mind and `HUGINN_ENDPOINT` (`rudp://host:port`) the daemon that +//! holds it. Each call is a new process; there is no retry inside it, the +//! caller's rule retries `Unavailable` once. +//! +//! Answers go to stdout and the exit code says what the answer was: 0 answered, +//! 1 the mind said no (a refusal, a conflict, an absent document). Troubles go +//! to stderr as `: `: 2 the call is wrong, 3 Unavailable, 4 no +//! answer and a retry cannot help, 5 the read-back after a committed admission +//! differs from what was sent. + +mod config; +mod door; +mod recipes; +mod render; +mod schema; +mod trouble; + +use std::io::Write; +use std::process::ExitCode; +use std::sync::OnceLock; + +use clap::{CommandFactory, FromArgMatches, Parser, Subcommand}; +use cultnet_rs::{PROJECTION_DOCUMENT, Selection}; +use huginn_mind::eureka_pipeline::{Line, PipelineDocument, PipelineRef, Short, Slug, pipeline_key}; +use huginn_mind::{ + BATCH_MAX, Faculty, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionBatch, PipelineAdmissionOutcome, + PipelinePageItems, PipelineProvenance, PipelineSelectionPage, SemanticQuery, +}; +use serde_json::Value; + +use config::{Settings, TOOL}; +use recipes::Recipe; +use trouble::Trouble; + +/// The page size when a selection names none. +const DEFAULT_LIMIT: u32 = 50; +/// The neighbours `--near` asks for when `--top-k` is not given. +const DEFAULT_TOP_K: u32 = 10; + +const ADMIT_HELP: &str = "\ +The file is a JSON array of 1 to 64 documents, each {\"kind\": , \"value\": {...}}, or - for stdin; one document object alone is a batch of one. \ +`huginn schema` lists the kinds and `huginn schema ` prints a value's JSON Schema. \ +Every key the schema does not know is refused (exit 2) with its path; nothing is sent. \ +After Committed or AlreadyAdmitted the batch is read back and compared with what was sent: `read back equal`, or exit 5 with the ids and field paths that differ."; + +#[derive(Parser, Debug)] +#[command(name = "huginn", about = "Huginn's mind: whoami, view, query, recipes, admit, schema", arg_required_else_help = true)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand, Debug)] +enum Command { + /// Who this process acts for and whether the daemon answers. + Whoami, + /// Read one document by id, with its admission facts and status. + View { + /// The document's id: ::. + id: String, + /// Print only the {kind, value} JSON, which `admit` accepts back. + #[arg(long)] + raw: bool, + }, + /// Query the mind with a CultNet selection. + Query { + /// The selection JSON, @, or - for stdin. + selection: String, + /// Rank by nearness to this text instead of ordering by key. + #[arg(long)] + near: Option, + /// How many neighbours (default 10). + #[arg(long, requires = "near")] + top_k: Option, + /// Print each item as `view` does. + #[arg(long)] + documents: bool, + }, + #[command(flatten)] + Recipe(Recipe), + /// Admit a batch of documents, whole or not at all, and read it back. + #[command(after_help = ADMIT_HELP)] + Admit { + #[arg(long, value_parser = faculty)] + faculty: Faculty, + /// The admitting agent's name. + #[arg(long)] + agent: String, + /// The admitting session. + #[arg(long)] + session: String, + /// The documents file, or - for stdin. + file: String, + }, + /// The JSON Schema of a document kind or of a selection; no name lists them. + Schema { name: Option }, +} + +fn faculty(text: &str) -> Result { + serde_json::from_value(Value::String(text.to_string())).map_err(|error| error.to_string()) +} + +/// The version, then the sha the binary was built from. +fn version() -> &'static str { + static VERSION: OnceLock = OnceLock::new(); + VERSION.get_or_init(|| format!("{} ({})", env!("CARGO_PKG_VERSION"), build_sha())) +} + +fn build_sha() -> &'static str { + option_env!("HUGINN_BUILD_SHA").unwrap_or("unknown") +} + +/// The command line, or the exit code of having answered it (help, version) or +/// refused it (usage, as `InvalidInput`). +fn parse() -> Result { + let mut command = Cli::command().version(version()); + for (name, text) in recipes::help() { + command = command.mut_subcommand(name, |recipe| recipe.after_help(text)); + } + match command.try_get_matches() { + Ok(matches) => Cli::from_arg_matches(&matches).map_err(|error| usage(&error)), + Err(error) => Err(usage(&error)), + } +} + +fn usage(error: &clap::Error) -> u8 { + if error.use_stderr() { + let text = error.to_string(); + eprintln!("InvalidInput: {}", text.trim_start_matches("error: ").trim_end()); + 2 + } else { + let _ = error.print(); + 0 + } +} + +fn main() -> ExitCode { + let cli = match parse() { + Ok(cli) => cli, + Err(code) => return ExitCode::from(code), + }; + match run(&Settings::from_env(), cli.command) { + Ok(code) => ExitCode::from(code), + Err(trouble) => { + eprintln!("{trouble}"); + ExitCode::from(trouble.exit()) + } + } +} + +fn say(text: impl std::fmt::Display) { + let _ = writeln!(std::io::stdout(), "{text}"); +} + +fn run(settings: &Settings, command: Command) -> Result { + match command { + Command::Whoami => whoami(settings), + Command::View { id, raw } => view(settings, &id, raw), + Command::Query { selection, near, top_k, documents } => { + let selection = door::selection(&door::argument(&selection)?)?; + let semantic = near.map(|text| SemanticQuery { text: Line(text), top_k: top_k.unwrap_or(DEFAULT_TOP_K) }); + query(settings, selection, semantic, None, documents) + } + Command::Recipe(recipe) => { + let ask = recipe.ask()?; + query(settings, door::selection_value(&ask.selection)?, ask.semantic, ask.page.cursor, ask.page.documents) + } + Command::Admit { faculty, agent, session, file } => admit(settings, faculty, agent, session, &file), + Command::Schema { name } => { + let text = schema::text(name.as_deref())?; + let _ = write!(std::io::stdout(), "{text}"); + Ok(0) + } + } +} + +fn whoami(settings: &Settings) -> Result { + say(format_args!("instance {}", settings.instance.as_deref().unwrap_or("unset"))); + say(format_args!("endpoint {}", settings.endpoint.as_deref().unwrap_or("unset"))); + say(format_args!("client {}", build_sha())); + match settings.ask(|_| HuginnMindRequest::Whoami) { + Ok(HuginnMindResponse::Whoami(status)) => { + say("reachable yes"); + say(format_args!( + "documents {} receipts {} index {} epoch {}", + status.documents, + status.receipts, + render::index_state(&status.index), + status.schema_epoch + )); + Ok(0) + } + Ok(other) => { + say("reachable no"); + Err(unexpected(&other)) + } + Err(trouble) => { + say("reachable no"); + Err(trouble) + } + } +} + +fn view(settings: &Settings, id: &str, raw: bool) -> Result { + let reference = PipelineRef { kind: door::kind_of(id)?, id: Short(id.to_string()) }; + match settings.ask(|instance| HuginnMindRequest::View { instance: instance.clone(), id: reference })? { + HuginnMindResponse::View(Some(found)) => { + say(render::view(&found, raw)); + Ok(0) + } + HuginnMindResponse::View(None) => { + say(format_args!("absent {id}")); + Ok(1) + } + HuginnMindResponse::Refused(refusal) => Ok(refused(&refusal)), + other => Err(unexpected(&other)), + } +} + +fn query( + settings: &Settings, + mut selection: Selection, + semantic: Option, + cursor: Option, + documents: bool, +) -> Result { + if documents { + selection.projection = PROJECTION_DOCUMENT.to_string(); + } + if cursor.is_some() { + selection.cursor = cursor; + } + selection.limit.get_or_insert(DEFAULT_LIMIT); + match settings.ask(|instance| HuginnMindRequest::Query { instance: instance.clone(), selection, semantic })? { + HuginnMindResponse::Query(page) => { + show(&page); + Ok(0) + } + HuginnMindResponse::Refused(refusal) => Ok(refused(&refusal)), + other => Err(unexpected(&other)), + } +} + +fn show(page: &PipelineSelectionPage) { + match &page.items { + PipelinePageItems::Headers(headers) => { + for header in headers { + say(render::header(header)); + } + } + PipelinePageItems::Documents(views) => { + for (index, found) in views.iter().enumerate() { + if index > 0 { + say(""); + } + say(render::view(found, false)); + } + } + } + say(format_args!("matched {} as_of {} next {}", page.matched, page.as_of, page.next.as_deref().unwrap_or("-"))); +} + +fn admit(settings: &Settings, faculty: Faculty, agent: String, session: String, file: &str) -> Result { + let documents = door::documents(&door::file_or_stdin(file)?)?; + if documents.is_empty() || documents.len() > BATCH_MAX { + return Err(Trouble::invalid(format!("a batch is 1 to {BATCH_MAX} documents, this one has {}", documents.len()))); + } + let sent = documents.clone(); + let provenance = PipelineProvenance { faculty, agent: Short(agent), session: Short(session), tool: Short(TOOL.to_string()) }; + let request = |instance: &Slug| { + HuginnMindRequest::Admit(PipelineAdmissionBatch { instance: instance.clone(), provenance, documents }) + }; + match settings.ask(request)? { + HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { receipt_id, committed_at, writes }) => { + say(format_args!("Committed {receipt_id} at {committed_at}")); + for write in &writes { + say(format_args!("wrote {}", write.id.0)); + } + read_back(settings, &sent) + } + HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id }) => { + say(format_args!("AlreadyAdmitted {receipt_id}")); + read_back(settings, &sent) + } + HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(refusal)) | HuginnMindResponse::Refused(refusal) => { + Ok(refused(&refusal)) + } + HuginnMindResponse::Admit(PipelineAdmissionOutcome::Conflict { identities }) => { + say("Conflict"); + for identity in &identities { + say(format_args!("conflicts {}", identity.id.0)); + } + Ok(1) + } + other => Err(unexpected(&other)), + } +} + +/// The end-to-end check that the mind kept what was sent: one `keys` query, +/// each stored document compared with the one sent. A difference is exit 5, +/// not silence, whatever dropped it (an older leaf, a serde default, skew). +fn read_back(settings: &Settings, sent: &[PipelineDocument]) -> Result { + let keys = sent + .iter() + .map(|document| pipeline_key(document).map_err(|refusal| Trouble::Internal(format!("a committed document has no key: {refusal:?}")))) + .collect::, _>>()?; + let selection = Selection { + keys: Some(keys.clone()), + projection: PROJECTION_DOCUMENT.to_string(), + limit: Some(u32::try_from(keys.len()).unwrap_or(u32::MAX)), + ..Selection::default() + }; + let page = match settings.ask(|instance| HuginnMindRequest::Query { instance: instance.clone(), selection, semantic: None })? { + HuginnMindResponse::Query(page) => page, + HuginnMindResponse::Refused(refusal) => return Ok(refused(&refusal)), + other => return Err(unexpected(&other)), + }; + let PipelinePageItems::Documents(stored) = page.items else { + return Err(Trouble::Internal("the read-back answered headers, not documents".to_string())); + }; + let mut differing = Vec::new(); + for (key, document) in keys.iter().zip(sent) { + let held = stored.iter().find(|found| &found.id.id.0 == key); + match held { + None => differing.push(format!("{key}: missing from the read-back")), + Some(found) if found.document == *document => {} + Some(found) => { + let (mut paths, sent_value, held_value) = (Vec::new(), to_value(document), to_value(&found.document)); + differences(&sent_value, &held_value, "", &mut paths); + let shown = if paths.is_empty() { "the document".to_string() } else { paths.join(", ") }; + differing.push(format!("{key}: {shown}")); + } + } + } + if differing.is_empty() { + say(format_args!("read back {} equal", sent.len())); + Ok(0) + } else { + Err(Trouble::ReadBackDiffers(format!( + "{}; the batch is committed, so supersede the documents that differ", + differing.join("; ") + ))) + } +} + +fn to_value(document: &PipelineDocument) -> Value { + serde_json::to_value(document).unwrap_or(Value::Null) +} + +/// The field paths at which two documents differ, including a field only one +/// of them carries. +fn differences(sent: &Value, held: &Value, path: &str, out: &mut Vec) { + match (sent, held) { + (Value::Object(sent), Value::Object(held)) => { + for key in sent.keys().chain(held.keys().filter(|key| !sent.contains_key(*key))) { + let here = if path.is_empty() { key.clone() } else { format!("{path}.{key}") }; + match (sent.get(key), held.get(key)) { + (Some(sent), Some(held)) => differences(sent, held, &here, out), + _ => out.push(here), + } + } + } + (Value::Array(sent), Value::Array(held)) if sent.len() == held.len() => { + for (index, (sent, held)) in sent.iter().zip(held).enumerate() { + differences(sent, held, &format!("{path}[{index}]"), out); + } + } + _ if sent == held => {} + _ => out.push(path.to_string()), + } +} + +/// A refusal is the mind's answer: stdout, exit 1. +fn refused(refusal: &MindRefusal) -> u8 { + say(format_args!("Refused {refusal:?}")); + 1 +} + +fn unexpected(answer: &HuginnMindResponse) -> Trouble { + let name = match answer { + HuginnMindResponse::Whoami(_) => "whoami", + HuginnMindResponse::Admit(_) => "admit", + HuginnMindResponse::View(_) => "view", + HuginnMindResponse::Query(_) => "query", + HuginnMindResponse::Persona(_) => "a persona", + HuginnMindResponse::PersonaPut(_) => "a persona put", + HuginnMindResponse::Refused(_) => "a refusal", + HuginnMindResponse::Deferred(_) => "a deferral", + }; + Trouble::Internal(format!("the daemon answered with {name}, which this call did not ask for")) +} diff --git a/crates/eureka-state/src/bin/huginn/recipes.rs b/crates/eureka-state/src/bin/huginn/recipes.rs new file mode 100644 index 0000000..78bf092 --- /dev/null +++ b/crates/eureka-state/src/bin/huginn/recipes.rs @@ -0,0 +1,287 @@ +//! The campaign progress view as named subcommands. Each recipe is exactly the +//! selection `references/campaign-state.md` gives (Eureka 278b881, "Recipes"), +//! built per call and never cached. Each one's `--help` prints its selection. + +use clap::{Args, Subcommand}; +use huginn_mind::SemanticQuery; +use huginn_mind::eureka_pipeline::{Line, PipelineKind}; +use serde_json::{Value, json}; + +use crate::door; +use crate::trouble::Trouble; + +/// The campaign whose `gap-` follow-ups are the substrate's known gaps. +const GAP_HOME: &str = "eureka-substrate"; +/// How many neighbours `precedent` asks the index for. +const PRECEDENT_TOP_K: u32 = 10; +/// Severities that block a cut. +const BLOCKING: [&str; 2] = ["Blocker", "High"]; + +#[derive(Args, Clone, Debug, Default)] +pub struct Paging { + /// The `next` cursor of the previous page. + #[arg(long)] + pub cursor: Option, + /// Print each item as `view` does, not as a header line. + #[arg(long)] + pub documents: bool, +} + +#[derive(Subcommand, Debug)] +pub enum Recipe { + /// Every campaign. + Campaigns(Paging), + /// The target in force. + Target { + campaign: String, + #[command(flatten)] + page: Paging, + }, + /// Rulings in force, operator directions included. + Rulings { + #[arg(required = true)] + campaigns: Vec, + #[command(flatten)] + page: Paging, + }, + /// Open questions and follow-ups. + Open { + #[arg(required = true)] + campaigns: Vec, + #[command(flatten)] + page: Paging, + }, + /// Open findings admitted by Soul. + Findings { + campaign: String, + /// Only Blocker and High. + #[arg(long)] + blocking: bool, + #[command(flatten)] + page: Paging, + }, + /// Specs with no report. + Unreported { + campaign: String, + #[command(flatten)] + page: Paging, + }, + /// Open questions raised in a spec: the spec waits on the operator. + Blocked { + spec_id: String, + #[command(flatten)] + page: Paging, + }, + /// Reports with no verdict: Soul's queue. + Unjudged { + campaign: String, + #[command(flatten)] + page: Paging, + }, + /// A subject's resolutions, newest first. + History { + id: String, + #[command(flatten)] + page: Paging, + }, + /// One cut's spec, reports, verdicts, findings and resolutions. + Cut { + campaign: String, + label: String, + #[command(flatten)] + page: Paging, + }, + /// Every spec and report, whole. + Ledger { + campaign: String, + #[command(flatten)] + page: Paging, + }, + /// Rulings, resolutions and findings nearest a question in plain words. + Precedent { + campaign: String, + text: String, + #[command(flatten)] + page: Paging, + }, + /// The substrate's known gaps. + Gaps(Paging), +} + +/// What a recipe asks the mind: the selection, the semantic part when it has +/// one, and the caller's paging flags. +pub struct Ask { + pub selection: Value, + pub semantic: Option, + pub page: Paging, +} + +impl Recipe { + pub fn ask(self) -> Result { + let (selection, semantic, page) = match self { + Self::Campaigns(page) => (campaigns(), None, page), + Self::Target { campaign, page } => (target(&campaign), None, page), + Self::Rulings { campaigns, page } => (rulings(&campaigns), None, page), + Self::Open { campaigns, page } => (open(&campaigns), None, page), + Self::Findings { campaign, blocking, page } => (findings(&campaign, blocking), None, page), + Self::Unreported { campaign, page } => (unreported(&campaign), None, page), + Self::Blocked { spec_id, page } => (blocked(&spec_id), None, page), + Self::Unjudged { campaign, page } => (unjudged(&campaign), None, page), + Self::History { id, page } => (history(door::kind_of(&id)?.type_id(), &id), None, page), + Self::Cut { campaign, label, page } => (cut(&campaign, &label), None, page), + Self::Ledger { campaign, page } => (ledger(&campaign), None, page), + Self::Precedent { campaign, text, page } => { + (precedent(&campaign), Some(SemanticQuery { text: Line(text), top_k: PRECEDENT_TOP_K }), page) + } + Self::Gaps(page) => (gaps(), None, page), + }; + Ok(Ask { selection, semantic, page }) + } +} + +fn schema(kind: PipelineKind) -> &'static str { + kind.type_id() +} + +fn root(campaigns: &[String]) -> Value { + json!({ "index": "root", "op": "any_of", "values": campaigns }) +} + +fn in_force() -> Value { + json!({ "index": "in_force", "op": "any_of", "values": ["true"] }) +} + +fn severity_blocking() -> Value { + json!({ "index": "severity", "op": "any_of", "values": BLOCKING }) +} + +fn campaigns() -> Value { + json!({ "schemas": [schema(PipelineKind::Campaign)] }) +} + +fn target(campaign: &str) -> Value { + json!({ + "schemas": [schema(PipelineKind::Target)], + "fields": [root(&[campaign.to_string()]), in_force()], + "projection": "document", + }) +} + +fn rulings(campaigns: &[String]) -> Value { + json!({ "schemas": [schema(PipelineKind::Ruling)], "fields": [root(campaigns), in_force()] }) +} + +fn open(campaigns: &[String]) -> Value { + json!({ + "schemas": [schema(PipelineKind::Question), schema(PipelineKind::FollowUp)], + "fields": [root(campaigns), in_force()], + }) +} + +fn findings(campaign: &str, blocking: bool) -> Value { + let mut fields = vec![ + root(&[campaign.to_string()]), + in_force(), + json!({ "index": "faculty", "op": "any_of", "values": ["Soul"] }), + ]; + if blocking { + fields.push(severity_blocking()); + } + json!({ "schemas": [schema(PipelineKind::Finding)], "fields": fields }) +} + +fn unreported(campaign: &str) -> Value { + json!({ + "schemas": [schema(PipelineKind::CutSpec)], + "fields": [root(&[campaign.to_string()]), in_force()], + "cited": { "role": "cut_spec", "exists": false }, + }) +} + +fn blocked(spec_id: &str) -> Value { + json!({ + "schemas": [schema(PipelineKind::Question)], + "fields": [in_force()], + "cites": { "target": { "schemaId": schema(PipelineKind::CutSpec), "recordKey": spec_id }, "role": "raised_in" }, + }) +} + +fn unjudged(campaign: &str) -> Value { + json!({ + "schemas": [schema(PipelineKind::CutReport)], + "fields": [root(&[campaign.to_string()])], + "cited": { "role": "cut_report", "exists": false }, + }) +} + +fn history(schema_id: &str, id: &str) -> Value { + json!({ + "schemas": [schema(PipelineKind::Resolution)], + "cites": { "target": { "schemaId": schema_id, "recordKey": id }, "role": "subject" }, + "descending": true, + }) +} + +fn cut(campaign: &str, label: &str) -> Value { + json!({ + "schemas": [ + schema(PipelineKind::CutSpec), + schema(PipelineKind::CutReport), + schema(PipelineKind::Verdict), + schema(PipelineKind::Finding), + schema(PipelineKind::Resolution), + ], + "fields": [root(&[campaign.to_string()]), { "index": "cut", "op": "any_of", "values": [label] }], + }) +} + +fn ledger(campaign: &str) -> Value { + json!({ + "schemas": [schema(PipelineKind::CutSpec), schema(PipelineKind::CutReport)], + "fields": [root(&[campaign.to_string()])], + "projection": "document", + }) +} + +fn precedent(campaign: &str) -> Value { + json!({ + "schemas": [schema(PipelineKind::Ruling), schema(PipelineKind::Resolution), schema(PipelineKind::Finding)], + "fields": [root(&[campaign.to_string()])], + }) +} + +fn gaps() -> Value { + json!({ + "schemas": [schema(PipelineKind::FollowUp)], + "fields": [root(&[GAP_HOME.to_string()]), in_force()], + }) +} + +/// Each recipe's `--help` addendum: its selection with placeholders in the +/// arguments' places. The same builders as execution, so help cannot drift +/// from what runs. +pub fn help() -> Vec<(&'static str, String)> { + let c = || "".to_string(); + let selection = |value: Value| format!("selection:\n{}", serde_json::to_string_pretty(&value).unwrap_or_default()); + let findings_help = + format!("{}\n--blocking adds the field predicate:\n{}", selection(findings(&c(), false)), severity_blocking()); + let precedent_help = format!( + "{}\nsemantic: {{ \"text\": \"\", \"top_k\": {PRECEDENT_TOP_K} }}", + selection(precedent(&c())) + ); + vec![ + ("campaigns", selection(campaigns())), + ("target", selection(target(&c()))), + ("rulings", selection(rulings(&[c()]))), + ("open", selection(open(&[c()]))), + ("findings", findings_help), + ("unreported", selection(unreported(&c()))), + ("blocked", selection(blocked(""))), + ("unjudged", selection(unjudged(&c()))), + ("history", selection(history("epiphany.pipeline..v2", ""))), + ("cut", selection(cut(&c(), "