lifeloop-cli 0.5.0

//! `lifeloop event invoke` — drive the host lifecycle pipeline.
//!
//! Reads a [`DispatchEnvelope`] JSON document from stdin (carrying the
//! [`CallbackRequest`] alongside any opaque [`PayloadEnvelope`] bodies
//! to deliver), then runs:
//! 1. router validation + adapter resolution (`router::route`)
//! 2. capability/placement negotiation (`router::negotiate` with an
//!    empty default `CapabilityRequest` and the envelope's payloads)
//! 3. subprocess callback against `--client-cmd` (with optional
//!    `--client-arg`s and `--timeout-ms`); the same payloads flow into
//!    the invoker so subprocess clients can reach them through the
//!    transport-boundary [`DispatchEnvelope`].
//! 4. receipt synthesis (`LifeloopReceiptEmitter::synthesize_and_emit`)
//!
//! Emits the synthesized `LifecycleReceipt` JSON on stdout. Failure
//! paths (router error, subprocess error, receipt error) all become
//! validation-class CLI errors so callers can distinguish "ran but the
//! contract said no" from "I couldn't read your input".

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use lifeloop::router::{
    BuiltinAdapterRegistry, CapabilityRequest, LifeloopReceiptEmitter, ReceiptContext,
    SubprocessCallbackInvoker, SubprocessInvokerConfig, negotiate, route,
};
use lifeloop::{CallbackResponse, DispatchEnvelope};

use super::{CliError, parse_stdin_json, print_json};

/// Top-level entry: `lifeloop event <action> ...`.
pub fn run<I: Iterator<Item = String>>(mut args: I) -> Result<(), CliError> {
    let action = args
        .next()
        .ok_or_else(|| CliError::Usage("event requires a subcommand: invoke".to_string()))?;
    match action.as_str() {
        "invoke" => run_invoke(args),
        other => Err(CliError::Usage(format!(
            "event: unknown subcommand `{other}` (expected: invoke)"
        ))),
    }
}

#[derive(Debug)]
struct InvokeArgs {
    client_cmd: Option<String>,
    client_args: Vec<String>,
    timeout_ms: u64,
    client_id: String,
    receipt_id: String,
    at_epoch_s: u64,
    in_process: bool,
}

/// Working state during arg parsing: `receipt_id` and `at_epoch_s` are
/// optional here so the parser can distinguish "not supplied" from a
/// caller-supplied value, then synthesize / default them at the end. A
/// constant `receipt_id` ("rcpt-cli") or epoch-0 timestamp would make
/// every flagless invocation emit a non-unique receipt id and a 1970
/// timestamp, collapsing the `(at_epoch_s, receipt_id)` ordering key.
struct InvokeArgsBuilder {
    client_cmd: Option<String>,
    client_args: Vec<String>,
    timeout_ms: u64,
    client_id: String,
    receipt_id: Option<String>,
    at_epoch_s: Option<u64>,
    in_process: bool,
}

impl InvokeArgs {
    fn parse<I: Iterator<Item = String>>(mut args: I) -> Result<Self, CliError> {
        let mut parsed = InvokeArgsBuilder {
            client_cmd: None,
            client_args: Vec::new(),
            timeout_ms: 5_000,
            client_id: "lifeloop-cli".to_string(),
            receipt_id: None,
            at_epoch_s: None,
            in_process: false,
        };

        while let Some(arg) = args.next() {
            match arg.as_str() {
                "--client-cmd" => {
                    parsed.client_cmd = Some(require_value(&arg, args.next())?);
                }
                "--client-arg" => {
                    parsed.client_args.push(require_value(&arg, args.next())?);
                }
                "--timeout-ms" => {
                    let value = require_value(&arg, args.next())?;
                    parsed.timeout_ms = parse_u64_flag("--timeout-ms", &value)?;
                }
                "--client-id" => {
                    parsed.client_id = require_value(&arg, args.next())?;
                }
                "--receipt-id" => {
                    parsed.receipt_id = Some(require_value(&arg, args.next())?);
                }
                "--at-epoch-s" => {
                    let value = require_value(&arg, args.next())?;
                    parsed.at_epoch_s = Some(parse_u64_flag("--at-epoch-s", &value)?);
                }
                "--in-process" => {
                    parsed.in_process = true;
                }
                other => {
                    return Err(CliError::Usage(format!(
                        "event invoke: unknown flag `{other}`"
                    )));
                }
            }
        }

        if !parsed.in_process && parsed.client_cmd.is_none() {
            return Err(CliError::Usage(
                "event invoke: --client-cmd <path> is required (or pass --in-process)".into(),
            ));
        }

        // `receipt_id` must be a new opaque identifier per emission (spec
        // wire-surface table). When `--receipt-id` is omitted we
        // synthesize a unique one rather than reusing a constant, which
        // would collapse the receipt-id half of the ordering key.
        // `at_epoch_s` defaults to the real wall clock when unset so the
        // diagnostic ordering key is not pinned to 1970.
        let receipt_id = parsed.receipt_id.unwrap_or_else(synthesize_receipt_id);

        Ok(Self {
            client_cmd: parsed.client_cmd,
            client_args: parsed.client_args,
            timeout_ms: parsed.timeout_ms,
            client_id: parsed.client_id,
            receipt_id,
            at_epoch_s: parsed.at_epoch_s.unwrap_or_else(epoch_s),
            in_process: parsed.in_process,
        })
    }

    fn receipt_context(&self, request: &lifeloop::CallbackRequest) -> ReceiptContext {
        ReceiptContext {
            client_id: self.client_id.clone(),
            receipt_id: self.receipt_id.clone(),
            parent_receipt_id: None,
            at_epoch_s: self.at_epoch_s,
            harness_session_id: request.harness_session_id.clone(),
            harness_run_id: request.harness_run_id.clone(),
            harness_task_id: request.harness_task_id.clone(),
        }
    }

    fn invoke_client(
        &self,
        plan: &lifeloop::router::RoutingPlan,
        payloads: &[lifeloop::PayloadEnvelope],
    ) -> Result<CallbackResponse, CliError> {
        if self.in_process {
            let response = CallbackResponse::ok(lifeloop::ReceiptStatus::Delivered);
            response.validate().map_err(|e| {
                CliError::Validation(format!("in-process response failed validation: {e}"))
            })?;
            return Ok(response);
        }

        let mut config = SubprocessInvokerConfig::new(
            self.client_cmd.as_ref().expect("checked by parse"),
            Duration::from_millis(self.timeout_ms),
        );
        config = config.args(self.client_args.iter().cloned());
        let invoker = SubprocessCallbackInvoker::new(config);

        use lifeloop::router::CallbackInvoker;
        invoker
            .invoke(plan, payloads)
            .map_err(|e| CliError::Validation(format!("subprocess callback failed: {e}")))
    }
}

fn run_invoke<I: Iterator<Item = String>>(args: I) -> Result<(), CliError> {
    let args = InvokeArgs::parse(args)?;
    let envelope: DispatchEnvelope = parse_stdin_json("DispatchEnvelope")?;
    envelope
        .validate()
        .map_err(|e| CliError::Validation(format!("DispatchEnvelope failed validation: {e}")))?;

    let registry = BuiltinAdapterRegistry;
    let plan = route(&envelope.request, &registry)
        .map_err(|e| CliError::Validation(format!("router rejected request: {e}")))?;

    // Negotiate against an empty capability request: the CLI doesn't
    // know what the caller wants beyond what's on the envelope. The
    // dispatch envelope's payloads feed real placement decisions
    // (issue #22) — passing the empty slice here would silently skip
    // payload-bearing negotiation.
    let cap_request = CapabilityRequest::new();
    let negotiated = negotiate(&plan, &cap_request, envelope.payloads.as_slice());

    let response = if negotiated.blocks_dispatch() {
        // Synthesize a receipt for the blocked path so the caller still
        // gets a structured result rather than just an error. `Observed`
        // (not `Failed`) keeps the response internally valid: a `Failed`
        // status would require a `failure_class`, which `ok` leaves
        // `None` and `validate()` rejects. `derive_status` ignores this
        // status on the blocked path — the blocking `Unsupported` /
        // `RequiresOperator` outcome wins and supplies the real
        // failure_class — so the value never needs to carry one.
        CallbackResponse::ok(lifeloop::ReceiptStatus::Observed)
    } else {
        args.invoke_client(&plan, envelope.payloads.as_slice())?
    };

    let ctx = args.receipt_context(&envelope.request);
    let emitter = LifeloopReceiptEmitter::in_memory();
    let receipt = emitter
        .synthesize_and_emit(&negotiated, &response, &ctx)
        .map_err(|e| CliError::Validation(format!("receipt emission failed: {e}")))?;
    print_json(&receipt)
}

fn parse_u64_flag(flag: &str, value: &str) -> Result<u64, CliError> {
    value
        .parse::<u64>()
        .map_err(|e| CliError::Usage(format!("{flag} must be a non-negative integer: {e}")))
}
fn require_value(flag: &str, value: Option<String>) -> Result<String, CliError> {
    value.ok_or_else(|| CliError::Usage(format!("flag `{flag}` requires a value")))
}

/// Wall-clock seconds since the Unix epoch (default `at_epoch_s` when
/// `--at-epoch-s` is not supplied). Falls back to 0 only if the clock is
/// before the epoch, which never happens in practice.
fn epoch_s() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Synthesize an opaque, per-emission-unique `receipt_id` when
/// `--receipt-id` is omitted. Composed from pid + a process-local
/// monotonic counter + nanosecond timestamp so two emissions in the same
/// process never collide (the bare nanosecond clock can repeat under a
/// coarse source; the counter guarantees uniqueness).
fn synthesize_receipt_id() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let pid = std::process::id();
    let seq = COUNTER.fetch_add(1, Ordering::Relaxed);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!("rcpt-cli-{pid}-{seq}-{nanos}")
}

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

    #[test]
    fn synthesized_receipt_ids_are_unique_per_call() {
        // The whole point of the synthesis: two emissions in one process
        // must not collapse onto the same receipt id (the old constant
        // "rcpt-cli" default did exactly that).
        let a = synthesize_receipt_id();
        let b = synthesize_receipt_id();
        assert_ne!(a, b, "synthesized receipt ids must be unique: {a} == {b}");
        assert!(a.starts_with("rcpt-cli-"));
    }
}