supercode-harness 0.4.16

//! ORC-13 — the orchestrator's WRITE door, the one every controlled-tier noun
//! goes through when `--harness orchestrator` names it.
//!
//! Hermes's write door is `hermes cron …`; OpenClaw's is `openclaw cron …`
//! through its Gateway. The orchestrator's is its own package: the only writer
//! that keeps the folder's byte-stability and its residue rules is `save()`
//! inside `sdk/orchestrator` (`docs/ORCHESTRATOR-IR.md` §1 rule 2, §6). This
//! module is the uniform client of that writer, and it never touches the
//! folder itself — no file here is opened for writing, ever.
//!
//! Two doors, one protocol, one answer (`docs/ORCHESTRATOR-IR.md` §4.6):
//!
//! * **live** — when `<root>/orchestrator.lock` names a process that is alive
//!   AND `<root>/orchestrator.sock` accepts a connection, one JSON line goes
//!   over that socket:
//!
//!   ```text
//!   {"op":"jobs.create","args":{…},"profile":"coder"}
//!   {"ok":true,"result":{"ran":"created cron job …","job":{…}}}
//!   ```
//!
//!   The daemon dispatches the matching operator event on its own queue, so
//!   the write is ordered against inbound messages and ticks and the running
//!   loop's in-memory state and the folder can never disagree.
//! * **cold** — otherwise the package's own CLI runs the identical verb
//!   through the identical `applyOperator`:
//!
//!   ```text
//!   node <entry> jobs.create --root <home> --profile coder --json '{…}'
//!   ```
//!
//!   The entry is resolved exactly the way `supercode orchestrator start`
//!   resolves it ([`crate::orchestrator::daemon_entry`]), so the cold path and
//!   the daemon are always the same build of the same package.
//!
//! Writing through a LIVE daemon's socket rather than its folder is not a
//! nicety: a daemon holds the loaded state in memory and re-saves it on every
//! `save` effect, so a folder edited behind its back would be silently
//! overwritten on the next tick. The lease is what makes that choice
//! mechanical instead of a guess.
//!
//! The answer is the PACKAGE's, never supercode's: `{ok:false, error}` is
//! surfaced verbatim (it carries the reducer's own refusal line), and each
//! caller re-reads its row through the ORCH readers afterwards.

use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};

use serde_json::Value;

/// Environment variable overriding the `node` used for the cold path (tests).
pub const NODE_BIN_ENV: &str = "SUPERCODE_NODE_BIN";

/// The daemon's local socket inside one orchestrator home.
pub const SOCKET_FILE: &str = "orchestrator.sock";

/// Which door answered.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Door {
    /// The running daemon's Unix socket.
    Live,
    /// The package's own CLI, run as a subprocess.
    Cold,
}

impl Door {
    /// Uniform spelling used in narrations and outcomes.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Live => "live",
            Self::Cold => "cold",
        }
    }
}

/// One answered operator call.
#[derive(Debug, Clone)]
pub struct DoorAnswer {
    /// The exact door that was driven, narrated.
    pub ran: String,
    /// Which door it was.
    pub door: Door,
    /// The package's own `result` object.
    pub result: Value,
}

/// Why an operator call could not be answered.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DoorError {
    /// The package refused the verb; the message is ITS words.
    Refused(String),
    /// The door itself could not be driven (no entry, no node, a crash).
    Failed(String),
}

impl std::fmt::Display for DoorError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Refused(message) | Self::Failed(message) => formatter.write_str(message),
        }
    }
}

impl std::error::Error for DoorError {}

type Result<T> = std::result::Result<T, DoorError>;

/// The daemon socket path for one home.
pub fn socket_path(root: &Path) -> PathBuf {
    root.join(SOCKET_FILE)
}

/// Whether a LIVE daemon is serving this home right now.
///
/// Both halves must hold: a lease naming a process that is alive, and a socket
/// file beside it. A lease whose process is gone is stale and a socket left by
/// a killed daemon is dead, so either alone would send the write into a void.
pub fn daemon_is_live(root: &Path) -> bool {
    crate::orchestrator::live_lease(root).is_some() && socket_path(root).exists()
}

/// Perform one operator verb through whichever door this home publishes.
pub fn call(root: &Path, op: &str, args: &Value, profile: &str) -> Result<DoorAnswer> {
    if daemon_is_live(root) {
        match call_live(root, op, args, profile) {
            Ok(answer) => return Ok(answer),
            // A refusal is the package's answer and is final. Only a broken
            // socket falls through to the cold path — a daemon that died
            // between the lease check and the connect must not lose the write.
            Err(DoorError::Refused(message)) => return Err(DoorError::Refused(message)),
            Err(DoorError::Failed(_)) => {}
        }
    }
    call_cold(root, op, args, profile)
}

/// The narration for the live door: what went over the socket.
fn narrate_live(root: &Path, op: &str, args: &Value, profile: &str) -> String {
    format!(
        "{} {op} --profile {profile} --json {}",
        socket_path(root).display(),
        shell_quote(&args.to_string())
    )
}

#[cfg(unix)]
fn call_live(root: &Path, op: &str, args: &Value, profile: &str) -> Result<DoorAnswer> {
    use std::os::unix::net::UnixStream;

    let ran = narrate_live(root, op, args, profile);
    let path = socket_path(root);
    let mut stream = UnixStream::connect(&path).map_err(|error| {
        DoorError::Failed(format!(
            "the orchestrator daemon is leased for `{}` but its socket `{}` did not accept a \
             connection: {error}",
            root.display(),
            path.display()
        ))
    })?;
    let line = serde_json::json!({"op": op, "args": args, "profile": profile});
    stream
        .write_all(format!("{line}\n").as_bytes())
        .and_then(|()| stream.flush())
        .map_err(|error| DoorError::Failed(format!("`{ran}` could not be sent: {error}")))?;
    let mut reader = BufReader::new(stream);
    let mut answer = String::new();
    reader
        .read_line(&mut answer)
        .map_err(|error| DoorError::Failed(format!("`{ran}` was not answered: {error}")))?;
    if answer.trim().is_empty() {
        return Err(DoorError::Failed(format!(
            "`{ran}`: the orchestrator closed the connection without answering"
        )));
    }
    interpret(&ran, Door::Live, answer.trim())
}

#[cfg(not(unix))]
fn call_live(root: &Path, op: &str, args: &Value, profile: &str) -> Result<DoorAnswer> {
    Err(DoorError::Failed(format!(
        "`{}`: the daemon's door is a Unix socket, which this platform has no client for; the \
         cold path answers instead",
        narrate_live(root, op, args, profile)
    )))
}

/// The package's own CLI: `node <entry> <op> --root … --profile … --json …`.
fn call_cold(root: &Path, op: &str, args: &Value, profile: &str) -> Result<DoorAnswer> {
    let entry = crate::orchestrator::daemon_entry().map_err(|error| {
        DoorError::Failed(format!(
            "the orchestrator's write door is its own package, and it could not be located: \
             {error}"
        ))
    })?;
    let node = std::env::var_os(NODE_BIN_ENV)
        .map(|value| value.to_string_lossy().trim().to_string())
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| "node".to_string());
    let payload = args.to_string();
    let arguments = vec![
        entry.to_string_lossy().into_owned(),
        op.to_string(),
        "--root".to_string(),
        root.to_string_lossy().into_owned(),
        "--profile".to_string(),
        profile.to_string(),
        "--json".to_string(),
        payload,
    ];
    let ran = std::iter::once(node.clone())
        .chain(arguments.iter().cloned())
        .map(|part| shell_quote(&part))
        .collect::<Vec<_>>()
        .join(" ");
    let output = std::process::Command::new(&node)
        .args(&arguments)
        .stdin(std::process::Stdio::null())
        .output()
        .map_err(|error| DoorError::Failed(format!("`{ran}` could not be executed: {error}")))?;
    let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
    let last = stdout.lines().rev().find(|line| !line.trim().is_empty());
    let Some(last) = last else {
        let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
        return Err(DoorError::Failed(format!(
            "`{ran}` printed nothing ({}){}",
            output.status,
            if stderr.is_empty() {
                String::new()
            } else {
                format!(": {stderr}")
            }
        )));
    };
    interpret(&ran, Door::Cold, last.trim())
}

/// One `{ok, result}` / `{ok:false, error}` line, whichever door printed it.
fn interpret(ran: &str, door: Door, line: &str) -> Result<DoorAnswer> {
    let value: Value = serde_json::from_str(line).map_err(|error| {
        DoorError::Failed(format!(
            "`{ran}` answered something that is not JSON: {error}"
        ))
    })?;
    if value.get("ok").and_then(Value::as_bool) == Some(true) {
        return Ok(DoorAnswer {
            ran: ran.to_string(),
            door,
            result: value.get("result").cloned().unwrap_or(Value::Null),
        });
    }
    // The package's own sentence, verbatim: a verb its reducer refused says
    // WHY, and that reason is the whole point of the controlled tier.
    Err(DoorError::Refused(
        value
            .get("error")
            .and_then(Value::as_str)
            .map(str::to_string)
            .unwrap_or_else(|| format!("`{ran}` answered `{line}`")),
    ))
}

/// The same narration quoting the rest of the tier uses.
fn shell_quote(value: &str) -> String {
    if !value.is_empty()
        && value
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || "-_./:@=+,".contains(c))
    {
        return value.to_string();
    }
    format!("'{}'", value.replace('\'', "'\\''"))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn scratch(label: &str) -> PathBuf {
        let root = std::env::temp_dir().join(format!(
            "supercode-orc13-{label}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&root).unwrap();
        root
    }

    /// A home with no lease and no socket is never "live": the door must fall
    /// through to the cold path rather than dial a socket that is not there.
    #[test]
    fn a_home_without_a_live_lease_is_not_live() {
        let root = scratch("live");
        assert!(!daemon_is_live(&root));
        // A lease alone is not enough — the socket has to be there too.
        crate::orchestrator::write_lease(
            &root,
            &crate::orchestrator::Lease {
                pid: std::process::id(),
                started_at: "2026-09-04T00:00:00Z".into(),
                root: root.clone(),
            },
        )
        .unwrap();
        assert!(!daemon_is_live(&root), "a lease without a socket is not up");
        std::fs::write(socket_path(&root), b"").unwrap();
        assert!(daemon_is_live(&root));
        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn a_refusal_carries_the_packages_own_sentence() {
        let error = interpret(
            "node entry jobs.delete",
            Door::Cold,
            r#"{"ok":false,"error":"jobs_delete: no job job_x"}"#,
        )
        .unwrap_err();
        assert_eq!(
            error,
            DoorError::Refused("jobs_delete: no job job_x".into())
        );
    }

    #[test]
    fn an_ok_line_yields_the_packages_result() {
        let answer = interpret(
            "node entry jobs.create",
            Door::Cold,
            r#"{"ok":true,"result":{"ran":"created cron job a","job_id":"a"}}"#,
        )
        .unwrap();
        assert_eq!(answer.door, Door::Cold);
        assert_eq!(
            answer.result.pointer("/job_id").and_then(Value::as_str),
            Some("a")
        );
    }

    /// The cold path is the package's own CLI, so a home that does not exist
    /// fails through the PACKAGE's loader, never through a Rust file write.
    #[test]
    fn the_cold_path_runs_the_packages_cli_and_refuses_a_home_that_does_not_load() {
        let root = scratch("cold").join("not-a-home");
        let error = call(
            &root,
            "jobs.delete",
            &serde_json::json!({"id": "x"}),
            "default",
        )
        .unwrap_err();
        let message = error.to_string();
        assert!(
            message.contains("not a directory") || message.contains("could not be executed"),
            "{message}"
        );
        std::fs::remove_dir_all(root.parent().unwrap()).ok();
    }
}