supercode-harness 0.5.45

//! 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: `<root>/orchestrator.sock`, or
/// `SUPERCODE_ORCHESTRATOR_SOCK` where the embedding keeps the socket off the
/// root (a home on a network filesystem cannot hold a Unix socket). The
/// daemon's loop derives the same path (`socketPathFor`).
pub fn socket_path(root: &Path) -> PathBuf {
    match std::env::var_os("SUPERCODE_ORCHESTRATOR_SOCK") {
        Some(path) if !path.is_empty() => PathBuf::from(path),
        _ => 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('\'', "'\\''"))
}