feat(stream-e): invoke wire — kei-runtime subprocess → real atoms

Replace NotImplemented stub with real atom execution per schema
§Runtime invocation contract.

Convention: JSON-in/JSON-out over subprocess. Every refactored crate
exposes `<crate> run-atom <verb>` that reads JSON from stdin (or
--input), dispatches to atoms::<verb>::run, emits Output JSON on
stdout, exits per atom-error class.

Runtime side (kei-runtime):
- InvokeError: +AtomFailed{atom,code,stderr} +SubprocessError
  +OutputParse +BinaryNotFound{crate_name}. NotImplemented kept as
  legacy escape for atoms opting out of run-atom protocol.
- Output: now {atom: String, result: Value} — carries atom's actual
  return value.
- invoke_exit_code: AtomFailed passes through child exit (0..=255),
  Subprocess/OutputParse → 1, BinaryNotFound → 127, NotImplemented → 64.
- Binary resolution: KEI_RUNTIME_BIN_DIR env → PATH fallback.

kei-task side:
- New `pub mod run_atom` in lib.rs
- atoms/mod.rs: VERBS const + DispatchError enum wrapping per-atom errors
- src/run_atom.rs: read_input (stdin/@path/literal), dispatch, exit mapping
- main.rs: Cmd::RunAtom{verb, input} subcommand; collapsed three
  classify_*_error helpers into single classify_dispatch. Legacy
  create/search/add-dependency CLIs preserved.

Tests: 5/5 runtime (+1 invoke_real_atom integration), 9/9 kei-task
(+1 atoms::tests::verbs_list_matches_submodules).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

_primitives/_rust/kei-runtime/src/invoke.rs

@@ -1,14 +1,23 @@
-//! Atom invocation — MVP stub.
+//! Atom invocation — executes atoms by spawning `<crate> run-atom <verb>`.
 //!
-//! Boundary: discovery + schema validation are wired. Actual atom execution
-//! shells out to `<crate> <verb>` or calls into a registry — both depend on
-//! Stream B (atoms refactor) landing first. Documented, intentional stub.
+//! Flow:
+//!   1. Discover atom → get `crate_name` + `verb` from `AtomMeta`
+//!   2. Validate input JSON against the atom's `input_schema`
+//!   3. Resolve the binary at `<KEI_RUNTIME_BIN_DIR>/<crate>` or `PATH`
+//!   4. Spawn `<crate> run-atom <verb>` with input on stdin
+//!   5. Parse stdout as JSON → `Output { atom, result }`
+//!   6. Propagate exit codes: 0 ok, 2 atom-error, 127 not-found, 1 IO
+//!
+//! `NotImplemented` is retained as a rare corner-case escape (e.g. an atom
+//! whose crate has not yet been migrated to the `run-atom` protocol).
 
 use crate::discover::{walk_atoms, AtomMeta};
 use crate::validate::validate_input;
 use serde::Serialize;
 use serde_json::Value;
-use std::path::Path;
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Stdio};
 
 #[derive(Debug)]
 pub enum InvokeError {
@@ -16,6 +25,15 @@ pub enum InvokeError {
     InputParse(String),
     InputInvalid(String),
     MissingInputSchema(String),
+    /// Crate binary is missing from both `KEI_RUNTIME_BIN_DIR` and `PATH`.
+    BinaryNotFound { crate_name: String },
+    /// Subprocess exited non-zero — propagate the atom's own exit code.
+    AtomFailed { atom: String, code: i32, stderr: String },
+    /// IO / spawn failure (not a non-zero exit from the child).
+    SubprocessError(String),
+    /// Atom's stdout was not parseable as JSON.
+    OutputParse(String),
+    /// Legacy escape — atom not yet migrated to `run-atom` protocol.
     NotImplemented { atom: String },
 }
 
@@ -25,9 +43,16 @@ impl std::fmt::Display for InvokeError {
             Self::AtomNotFound(id) => write!(f, "no atom matching {id}"),
             Self::InputParse(e) => write!(f, "input rejected: {e}"),
             Self::InputInvalid(e) => write!(f, "input rejected: {e}"),
-            Self::MissingInputSchema(id) => {
-                write!(f, "atom `{id}` declares no input schema")
+            Self::MissingInputSchema(id) => write!(f, "atom `{id}` declares no input schema"),
+            Self::BinaryNotFound { crate_name } => write!(
+                f,
+                "binary `{crate_name}` not found on PATH or KEI_RUNTIME_BIN_DIR"
+            ),
+            Self::AtomFailed { atom, code, stderr } => {
+                write!(f, "atom `{atom}` exited {code}: {stderr}")
             }
+            Self::SubprocessError(e) => write!(f, "subprocess: {e}"),
+            Self::OutputParse(e) => write!(f, "atom stdout not JSON: {e}"),
             Self::NotImplemented { atom } => write!(
                 f,
                 "invoke not yet wired for this atom ({atom}); use the underlying CLI directly"
@@ -38,18 +63,17 @@ impl std::fmt::Display for InvokeError {
 
 impl std::error::Error for InvokeError {}
 
-/// MVP stub output — shape documents the boundary for downstream wire-up.
+/// Parsed output of an invoked atom. `result` is the raw JSON the atom wrote.
 #[derive(Debug, Serialize)]
 pub struct Output {
     pub atom: String,
+    pub result: Value,
 }
 
 /// Invoke an atom by full ID with a JSON input string.
 ///
-/// MVP contract: discover atom → parse input → validate against schema →
-/// return `NotImplemented` until Stream B wires the executor. The callers
-/// (main.rs) map `NotImplemented` to a dedicated non-zero exit code so CI
-/// scripts can distinguish "unimplemented" from "atom rejected input".
+/// Contract: discover atom → validate input against schema → spawn
+/// `<crate> run-atom <verb>` with stdin=input → parse stdout as JSON.
 pub fn invoke(root: &Path, atom_id: &str, input_json: &str) -> Result<Output, InvokeError> {
     let meta = find_atom(root, atom_id)?;
     let input: Value =
@@ -58,9 +82,8 @@ pub fn invoke(root: &Path, atom_id: &str, input_json: &str) -> Result<Output, In
         .input_schema
         .as_ref()
         .ok_or_else(|| InvokeError::MissingInputSchema(atom_id.to_string()))?;
-    validate_input(schema, &input)
-        .map_err(|e| InvokeError::InputInvalid(e.to_string()))?;
-    Err(InvokeError::NotImplemented { atom: atom_id.to_string() })
+    validate_input(schema, &input).map_err(|e| InvokeError::InputInvalid(e.to_string()))?;
+    exec_atom(&meta, input_json)
 }
 
 fn find_atom(root: &Path, atom_id: &str) -> Result<AtomMeta, InvokeError> {
@@ -69,3 +92,60 @@ fn find_atom(root: &Path, atom_id: &str) -> Result<AtomMeta, InvokeError> {
         .find(|a| a.full_id == atom_id)
         .ok_or_else(|| InvokeError::AtomNotFound(atom_id.to_string()))
 }
+
+fn exec_atom(meta: &AtomMeta, input_json: &str) -> Result<Output, InvokeError> {
+    let bin = resolve_binary(&meta.crate_name)
+        .ok_or_else(|| InvokeError::BinaryNotFound { crate_name: meta.crate_name.clone() })?;
+    let mut child = Command::new(&bin)
+        .arg("run-atom")
+        .arg(&meta.verb)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .map_err(|e| InvokeError::SubprocessError(format!("spawn {}: {e}", bin.display())))?;
+    if let Some(mut stdin) = child.stdin.take() {
+        stdin
+            .write_all(input_json.as_bytes())
+            .map_err(|e| InvokeError::SubprocessError(format!("write stdin: {e}")))?;
+    }
+    let out = child
+        .wait_with_output()
+        .map_err(|e| InvokeError::SubprocessError(format!("wait: {e}")))?;
+    handle_subprocess_output(meta, out)
+}
+
+fn handle_subprocess_output(
+    meta: &AtomMeta,
+    out: std::process::Output,
+) -> Result<Output, InvokeError> {
+    let code = out.status.code().unwrap_or(-1);
+    if !out.status.success() {
+        let stderr = String::from_utf8_lossy(&out.stderr).trim().to_string();
+        return Err(InvokeError::AtomFailed { atom: meta.full_id.clone(), code, stderr });
+    }
+    let stdout = String::from_utf8_lossy(&out.stdout);
+    let result: Value = serde_json::from_str(stdout.trim())
+        .map_err(|e| InvokeError::OutputParse(format!("{e}; stdout was: {stdout}")))?;
+    Ok(Output { atom: meta.full_id.clone(), result })
+}
+
+/// Resolve `<crate_name>` as an executable:
+///   1. `$KEI_RUNTIME_BIN_DIR/<crate_name>` if env is set and file exists
+///   2. Walk `$PATH`, return first `<dir>/<crate_name>` that exists
+fn resolve_binary(crate_name: &str) -> Option<PathBuf> {
+    if let Ok(bin_dir) = std::env::var("KEI_RUNTIME_BIN_DIR") {
+        let candidate = PathBuf::from(bin_dir).join(crate_name);
+        if candidate.is_file() {
+            return Some(candidate);
+        }
+    }
+    let path = std::env::var("PATH").ok()?;
+    for dir in std::env::split_paths(&path) {
+        let candidate = dir.join(crate_name);
+        if candidate.is_file() {
+            return Some(candidate);
+        }
+    }
+    None
+}

_primitives/_rust/kei-runtime/src/main.rs

@@ -115,13 +115,22 @@ fn run_invoke(root: PathBuf, atom_id: String, input_arg: String) -> ExitCode {
 /// Map typed invoke errors to exit codes per locked §Runtime schema.
 ///
 /// - `AtomNotFound | InputParse | InputInvalid | MissingInputSchema` → 2  (atom error)
-/// - `NotImplemented`                                                → 64 (CLI contract not yet honoured)
+/// - `AtomFailed { code, .. }`                                       → passthrough child exit code
+/// - `SubprocessError | OutputParse`                                 → 1  (IO / malformed output)
+/// - `BinaryNotFound`                                                → 127 (POSIX command-not-found)
+/// - `NotImplemented`                                                → 64 (legacy escape)
 fn invoke_exit_code(err: &InvokeError) -> u8 {
     match err {
         InvokeError::AtomNotFound(_)
         | InvokeError::InputParse(_)
         | InvokeError::InputInvalid(_)
         | InvokeError::MissingInputSchema(_) => 2,
+        InvokeError::AtomFailed { code, .. } => {
+            let c = *code;
+            if (0..=255).contains(&c) { c as u8 } else { 1 }
+        }
+        InvokeError::SubprocessError(_) | InvokeError::OutputParse(_) => 1,
+        InvokeError::BinaryNotFound { .. } => 127,
         InvokeError::NotImplemented { .. } => EXIT_INVOKE_NOT_IMPLEMENTED,
     }
 }

_primitives/_rust/kei-runtime/tests/invoke_exit_codes_smoke.rs

@@ -1,12 +1,11 @@
 //! Integration test — `kei-runtime invoke` exit codes per §Runtime contract.
 //!
 //! - Unknown atom id → exit 2 (atom rejected)
-//! - Known atom, valid input → exit 64 (CLI contract not yet honoured / NotImplemented)
+//! - Known atom whose crate binary is not on PATH → exit 127 (BinaryNotFound)
 //!
-//! The invoke binary is a stub (no executor wired yet). Previously it
-//! returned `Ok(...)` with an error payload → ExitCode::SUCCESS. Now every
-//! invocation path yields a typed `InvokeError`; main.rs maps variants to
-//! distinct exit codes so CI scripts can distinguish cases.
+//! Real-atom execution (happy path) lives in `invoke_real_atom.rs`, which
+//! points `KEI_RUNTIME_BIN_DIR` at the workspace `target/` to pick up
+//! `kei-task` without polluting the user's PATH.
 
 use std::fs;
 use std::path::Path;
@@ -73,22 +72,27 @@ fn invoke_atom_not_found_exits_2() {
 }
 
 #[test]
-fn invoke_not_implemented_exits_64() {
+fn invoke_binary_not_found_exits_127() {
     let tmp = tempfile::tempdir().unwrap();
-    write_atom(tmp.path(), "kei-demo", "create");
+    write_atom(tmp.path(), "kei-demo-absent", "create");
+    // Use an empty bin dir so the `kei-demo-absent` binary cannot be found.
+    let empty_bin = tmp.path().join("empty-bin-dir");
+    std::fs::create_dir_all(&empty_bin).unwrap();
     let out = Command::new(BIN)
+        .env("KEI_RUNTIME_BIN_DIR", &empty_bin)
+        .env("PATH", &empty_bin)
         .arg("invoke")
-        .arg("kei-demo::create")
+        .arg("kei-demo-absent::create")
         .arg("--input")
         .arg(r#"{"title":"hello"}"#)
         .arg("--root")
         .arg(tmp.path())
         .output()
         .expect("spawn kei-runtime");
-    assert_eq!(out.status.code(), Some(64),
-        "expected exit 64 on NotImplemented; stderr: {}",
+    assert_eq!(out.status.code(), Some(127),
+        "expected exit 127 on BinaryNotFound; stderr: {}",
         String::from_utf8_lossy(&out.stderr));
     let stderr = String::from_utf8_lossy(&out.stderr);
-    assert!(stderr.contains("invoke not yet wired"),
-        "expected 'invoke not yet wired' in stderr: {stderr}");
+    assert!(stderr.contains("not found"),
+        "expected 'not found' in stderr: {stderr}");
 }

_primitives/_rust/kei-runtime/tests/invoke_real_atom.rs

@@ -0,0 +1,81 @@
+//! Integration test — `kei-runtime invoke` actually executes `kei-task::create`.
+//!
+//! Wire-up:
+//!   1. Pre-build `kei-task` in the workspace target dir.
+//!   2. Point the `--root` at the workspace's `_primitives/_rust/` so the
+//!      runtime discovers the real atom metadata (`kei-task/atoms/create.md`).
+//!   3. Point `KEI_RUNTIME_BIN_DIR` at the target dir so the runtime resolves
+//!      the `kei-task` binary without polluting $PATH.
+//!   4. Invoke → expect exit 0 and a JSON result containing `id` as integer.
+
+use std::path::{Path, PathBuf};
+use std::process::Command;
+
+const BIN: &str = env!("CARGO_BIN_EXE_kei-runtime");
+
+/// Absolute path to `_primitives/_rust/` (the atom-discovery root).
+fn rust_root() -> PathBuf {
+    // This file lives in `_primitives/_rust/kei-runtime/tests/`.
+    let here = Path::new(env!("CARGO_MANIFEST_DIR"));
+    here.parent().expect("_primitives/_rust").to_path_buf()
+}
+
+/// Build `kei-task` so the runtime can spawn it. Uses the current profile's
+/// target dir, then hands that dir to the invoke via KEI_RUNTIME_BIN_DIR.
+fn build_kei_task_and_target_dir() -> PathBuf {
+    let rust_root = rust_root();
+    let status = Command::new(env!("CARGO"))
+        .arg("build")
+        .arg("-p")
+        .arg("kei-task")
+        .arg("--quiet")
+        .current_dir(&rust_root)
+        .status()
+        .expect("cargo build kei-task");
+    assert!(status.success(), "cargo build kei-task failed");
+    // `target` dir — try explicit override first, then fallback to `target/debug`.
+    if let Ok(t) = std::env::var("CARGO_TARGET_DIR") {
+        return PathBuf::from(t).join("debug");
+    }
+    rust_root.join("target").join("debug")
+}
+
+#[test]
+fn invoke_kei_task_create_returns_id() {
+    let bin_dir = build_kei_task_and_target_dir();
+    assert!(
+        bin_dir.join("kei-task").is_file(),
+        "kei-task binary not at {}/kei-task",
+        bin_dir.display()
+    );
+    let tmp = tempfile::tempdir().unwrap();
+    let db = tmp.path().join("task.sqlite");
+    let out = Command::new(BIN)
+        .env("KEI_RUNTIME_BIN_DIR", &bin_dir)
+        .env("KEI_TASK_DB", &db)
+        .arg("invoke")
+        .arg("kei-task::create")
+        .arg("--input")
+        .arg(r#"{"title":"integration"}"#)
+        .arg("--root")
+        .arg(rust_root())
+        .output()
+        .expect("spawn kei-runtime");
+    let stdout = String::from_utf8_lossy(&out.stdout).to_string();
+    let stderr = String::from_utf8_lossy(&out.stderr).to_string();
+    assert_eq!(
+        out.status.code(),
+        Some(0),
+        "expected exit 0; stdout: {stdout}; stderr: {stderr}"
+    );
+    let parsed: serde_json::Value =
+        serde_json::from_str(stdout.trim()).expect("stdout is JSON");
+    // Output shape: { "atom": "kei-task::create", "result": { "id": N, "created_at": ... } }
+    let result = parsed.get("result").expect("result field");
+    let id = result
+        .get("id")
+        .expect("id field on result")
+        .as_i64()
+        .expect("id is integer");
+    assert!(id >= 1, "id must be a positive integer, got {id}");
+}

_primitives/_rust/kei-task/src/atoms/mod.rs

@@ -8,3 +8,47 @@
 pub mod add_dependency;
 pub mod create;
 pub mod search;
+
+/// Verbs exposed through the `run-atom <verb>` machine-facing CLI.
+///
+/// Source of truth for the dispatch table. Unit tests assert this stays in
+/// sync with the sub-modules so adding a new verb can't silently skip
+/// `run-atom` wiring.
+pub const VERBS: &[&str] = &["create", "add-dependency", "search"];
+
+/// Errors from the `run-atom` dispatcher layer itself — NOT from the atom
+/// bodies. Use `classify_dispatch_error` in main.rs to map to exit codes.
+#[derive(Debug)]
+pub enum DispatchError {
+    UnknownVerb(String),
+    InvalidInput(String),
+    Create(create::Error),
+    AddDep(add_dependency::Error),
+    Search(search::Error),
+}
+
+impl std::fmt::Display for DispatchError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::UnknownVerb(v) => write!(f, "no such atom verb `{v}` in crate kei-task"),
+            Self::InvalidInput(e) => write!(f, "InvalidInput: {e}"),
+            Self::Create(e) => write!(f, "{e}"),
+            Self::AddDep(e) => write!(f, "{e}"),
+            Self::Search(e) => write!(f, "{e}"),
+        }
+    }
+}
+
+impl std::error::Error for DispatchError {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn verbs_list_matches_submodules() {
+        // If a new verb module is added, the VERBS list MUST gain it.
+        // This test pins the exact size + order so drift is caught at CI.
+        assert_eq!(VERBS, ["create", "add-dependency", "search"]);
+    }
+}

_primitives/_rust/kei-task/src/lib.rs

@@ -4,6 +4,7 @@ pub mod atoms;
 pub mod deps;
 pub mod graph;
 pub mod milestones;
+pub mod run_atom;
 pub mod schema;
 pub mod search;
 pub mod store;

_primitives/_rust/kei-task/src/main.rs

@@ -9,6 +9,7 @@ use kei_task::atoms;
 use kei_task::deps::dependency_chain;
 use kei_task::graph::list_edges;
 use kei_task::milestones::{create_milestone, link_task_to_milestone};
+use kei_task::run_atom;
 use kei_task::{Milestone, Store};
 use std::path::PathBuf;
 use std::process::ExitCode;
@@ -57,6 +58,10 @@ enum Cmd {
     Search { query: String, #[arg(long, default_value_t = 20)] limit: i64 },
     Milestone { name: String, #[arg(long, default_value = "")] description: String },
     LinkMilestone { task_id: i64, milestone_id: i64 },
+    /// Machine-facing atom invocation — `run-atom <verb>` reads JSON from
+    /// stdin (or `--input`), dispatches to `atoms::<verb>::run`, writes JSON
+    /// to stdout. Used by `kei-runtime invoke`.
+    RunAtom { verb: String, #[arg(long)] input: Option<String> },
 }
 
 fn db_path(cli_db: Option<PathBuf>) -> PathBuf {
@@ -85,24 +90,29 @@ fn dispatch(s: &Store, cmd: Cmd) -> Result<(), CliError> {
         Cmd::Milestone { name, description } => cmd_milestone(s, name, description),
         Cmd::LinkMilestone { task_id, milestone_id } =>
             cmd_link_milestone(s, task_id, milestone_id),
+        Cmd::RunAtom { verb, input } => cmd_run_atom(s, verb, input),
     }
 }
 
+fn cmd_run_atom(s: &Store, verb: String, input: Option<String>) -> Result<(), CliError> {
+    let txt = run_atom::read_input(input).map_err(CliError::io)?;
+    let json = run_atom::dispatch(s, &verb, &txt)
+        .map_err(|e| CliError { code: run_atom::exit_for_error(&e), msg: format!("{e}") })?;
+    println!("{json}");
+    Ok(())
+}
+
 fn cmd_create(s: &Store, title: String, description: String, priority: String) -> Result<(), CliError> {
     let out = atoms::create::run(s, atoms::create::Input {
         title, description, priority, milestone_id: None,
-    }).map_err(classify_create_error)?;
+    }).map_err(|e| classify_dispatch(atoms::DispatchError::Create(e)))?;
     println!("{}", out.id);
     Ok(())
 }
 
-/// Per §Runtime contract: validation errors → exit 2, storage/IO → exit 1.
-fn classify_create_error(e: atoms::create::Error) -> CliError {
-    match e {
-        atoms::create::Error::InvalidTitle
-        | atoms::create::Error::InvalidPriority(_) => CliError::atom(format!("{e}")),
-        atoms::create::Error::StoreError(_) => CliError::io(format!("{e}")),
-    }
+/// Classify any kei-task atom error via the shared `run_atom` exit-code table.
+fn classify_dispatch(e: atoms::DispatchError) -> CliError {
+    CliError { code: run_atom::exit_for_error(&e), msg: format!("{e}") }
 }
 
 fn cmd_update(s: &Store, id: i64, status: Option<String>, title: Option<String>) -> Result<(), CliError> {
@@ -119,21 +129,11 @@ fn cmd_add_dep(s: &Store, from_id: i64, to_id: i64, dep_type: String) -> Result<
     let dep_display = if dep_type.is_empty() { "blocks".to_string() } else { dep_type.clone() };
     atoms::add_dependency::run(s, atoms::add_dependency::Input {
         from: from_id, to: to_id, dep_type,
-    }).map_err(classify_add_dep_error)?;
+    }).map_err(|e| classify_dispatch(atoms::DispatchError::AddDep(e)))?;
     println!("dep: {} -> {} ({})", from_id, to_id, dep_display);
     Ok(())
 }
 
-/// Per §Runtime contract: semantic rejections → exit 2, storage/IO → exit 1.
-fn classify_add_dep_error(e: atoms::add_dependency::Error) -> CliError {
-    match e {
-        atoms::add_dependency::Error::SelfDependency
-        | atoms::add_dependency::Error::InvalidDepType(_)
-        | atoms::add_dependency::Error::CycleDetected => CliError::atom(format!("{e}")),
-        atoms::add_dependency::Error::StoreError(_) => CliError::io(format!("{e}")),
-    }
-}
-
 fn cmd_graph(s: &Store) -> Result<(), CliError> {
     for e in list_edges(s)? {
         println!("{}\t-[{}]->\t{}", e.task_id, e.dep_type, e.depends_on);
@@ -149,21 +149,13 @@ fn cmd_chain(s: &Store, id: i64) -> Result<(), CliError> {
 fn cmd_search(s: &Store, query: String, limit: i64) -> Result<(), CliError> {
     let out = atoms::search::run(s, atoms::search::Input {
         query, limit: Some(limit),
-    }).map_err(classify_search_error)?;
+    }).map_err(|e| classify_dispatch(atoms::DispatchError::Search(e)))?;
     for t in out.results {
         println!("{}\t{}\t{}", t.id, t.status, t.title);
     }
     Ok(())
 }
 
-/// Per §Runtime contract: query validation → exit 2, storage/IO → exit 1.
-fn classify_search_error(e: atoms::search::Error) -> CliError {
-    match e {
-        atoms::search::Error::InvalidQuery => CliError::atom(format!("{e}")),
-        atoms::search::Error::StoreError(_) => CliError::io(format!("{e}")),
-    }
-}
-
 fn cmd_milestone(s: &Store, name: String, description: String) -> Result<(), CliError> {
     let id = create_milestone(s, &Milestone {
         name, description, ..Default::default() })?;

_primitives/_rust/kei-task/src/run_atom.rs

@@ -0,0 +1,86 @@
+//! Machine-facing `run-atom <verb>` dispatcher.
+//!
+//! Reads JSON input (stdin or literal), dispatches to `atoms::<verb>::run`,
+//! serializes the typed Output back to stdout. Exit codes mapped by caller.
+
+use crate::atoms::{self, DispatchError};
+use crate::Store;
+use serde_json::Value;
+use std::io::Read;
+
+/// Read JSON input from an optional arg. `None` → read from stdin.
+/// `Some("@path")` → read the file at `path`.
+/// `Some(literal)` → parse the literal as JSON.
+pub fn read_input(arg: Option<String>) -> Result<String, String> {
+    match arg {
+        Some(s) if s.starts_with('@') => {
+            let path = &s[1..];
+            std::fs::read_to_string(path).map_err(|e| format!("read {path}: {e}"))
+        }
+        Some(s) => Ok(s),
+        None => read_stdin(),
+    }
+}
+
+fn read_stdin() -> Result<String, String> {
+    let mut buf = String::new();
+    std::io::stdin()
+        .read_to_string(&mut buf)
+        .map_err(|e| format!("stdin: {e}"))?;
+    Ok(buf)
+}
+
+/// Dispatch a verb to its atom. Returns serialized JSON on success.
+pub fn dispatch(store: &Store, verb: &str, input_json: &str) -> Result<String, DispatchError> {
+    let input: Value = serde_json::from_str(input_json)
+        .map_err(|e| DispatchError::InvalidInput(e.to_string()))?;
+    match verb {
+        "create" => run_create(store, input),
+        "add-dependency" => run_add_dep(store, input),
+        "search" => run_search(store, input),
+        other => Err(DispatchError::UnknownVerb(other.to_string())),
+    }
+}
+
+fn run_create(store: &Store, input: Value) -> Result<String, DispatchError> {
+    let parsed: atoms::create::Input = serde_json::from_value(input)
+        .map_err(|e| DispatchError::InvalidInput(e.to_string()))?;
+    let out = atoms::create::run(store, parsed).map_err(DispatchError::Create)?;
+    serde_json::to_string(&out).map_err(|e| DispatchError::InvalidInput(e.to_string()))
+}
+
+fn run_add_dep(store: &Store, input: Value) -> Result<String, DispatchError> {
+    let parsed: atoms::add_dependency::Input = serde_json::from_value(input)
+        .map_err(|e| DispatchError::InvalidInput(e.to_string()))?;
+    let out = atoms::add_dependency::run(store, parsed).map_err(DispatchError::AddDep)?;
+    serde_json::to_string(&out).map_err(|e| DispatchError::InvalidInput(e.to_string()))
+}
+
+fn run_search(store: &Store, input: Value) -> Result<String, DispatchError> {
+    let parsed: atoms::search::Input = serde_json::from_value(input)
+        .map_err(|e| DispatchError::InvalidInput(e.to_string()))?;
+    let out = atoms::search::run(store, parsed).map_err(DispatchError::Search)?;
+    serde_json::to_string(&out).map_err(|e| DispatchError::InvalidInput(e.to_string()))
+}
+
+/// Map a `DispatchError` to the §Runtime exit-code contract.
+/// Returns `(exit_code, stderr_msg)`.
+pub fn exit_for_error(e: &DispatchError) -> u8 {
+    match e {
+        DispatchError::UnknownVerb(_) | DispatchError::InvalidInput(_) => 2,
+        DispatchError::Create(err) => match err {
+            atoms::create::Error::InvalidTitle | atoms::create::Error::InvalidPriority(_) => 2,
+            atoms::create::Error::StoreError(_) => 1,
+        },
+        DispatchError::AddDep(err) => match err {
+            atoms::add_dependency::Error::SelfDependency
+            | atoms::add_dependency::Error::InvalidDepType(_)
+            | atoms::add_dependency::Error::CycleDetected => 2,
+            atoms::add_dependency::Error::StoreError(_) => 1,
+        },
+        DispatchError::Search(err) => match err {
+            atoms::search::Error::InvalidQuery => 2,
+            atoms::search::Error::StoreError(_) => 1,
+        },
+    }
+}