lifeloop-cli 0.2.0

//! End-to-end integration: Lifeloop drives a CCD-shaped callback
//! client over the process boundary (issue #8).
//!
//! This is the load-bearing proof that a CCD-style client can be
//! reached through Lifeloop's router stack without taking a Rust
//! dependency on it: the only contract is a JSON
//! [`lifeloop::CallbackRequest`] on stdin and a JSON
//! [`lifeloop::CallbackResponse`] on stdout. The fake client lives in
//! `src/bin/lifeloop-fake-ccd-client.rs` and is reached through the
//! cargo-provided `CARGO_BIN_EXE_lifeloop-fake-ccd-client` env var so
//! the path is portable across Linux and macOS.
//!
//! # Behavior corpus ownership
//!
//! Each assertion below is annotated as either **Lifeloop-owned**
//! (the contract or the router stack must guarantee it) or
//! **CCD-owned** (a real client implementation can vary the value).
//! The split mirrors the issue acceptance bullet — Lifeloop tests the
//! seam, not the client semantics.
//!
//! * `request.event` / `request.event_id` / `request.adapter_id` /
//!   `request.adapter_version` / `request.integration_mode` /
//!   `request.invocation_id` / `request.frame_context` /
//!   `request.payload_refs` / `request.idempotency_key` arriving on
//!   stdin → **Lifeloop-owned**: the router synthesizes the request
//!   from the validated [`lifeloop::router::RoutingPlan`] and the
//!   wire shape is pinned by `tests/wire_contract.rs`.
//! * The specific `payload_kind` (`ccd.instruction_frame`) and the
//!   payload `body` text returned by the fake → **CCD-owned**: a
//!   real client picks its own client-defined kinds and bodies.
//! * `response.status == Delivered` for `frame.opening` and
//!   `session.started` happy paths → **CCD-owned** behavior; the
//!   contract only requires that *some* valid status come back.
//! * `receipt.emitted` being rejected before spawn →
//!   **Lifeloop-owned**: the contract states `receipt.emitted` is a
//!   notification event and never produces a downstream invocation.
//! * Failure-class mapping for transport / parse / timeout / non-zero
//!   exit → **Lifeloop-owned**: the failure-class vocabulary is part
//!   of Lifeloop's surface and the router must map subprocess
//!   failures onto it deterministically.

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::time::Duration;

use lifeloop::router::{
    AdapterRegistry, AdapterResolution, CallbackInvoker, RoutingPlan, SubprocessCallbackInvoker,
    SubprocessInvokerConfig, SubprocessInvokerError, route,
};
use lifeloop::{
    AcceptablePlacement, AdapterManifest, AdapterRole, CallbackRequest, ConformanceLevel,
    FailureClass, FrameContext, IntegrationMode, LifecycleEventKind, ManifestContextPressure,
    ManifestPlacementClass, ManifestPlacementSupport, ManifestReceipts, PayloadEnvelope,
    PayloadRef, PlacementClass, ReceiptStatus, RegisteredAdapter, RequirementLevel, SCHEMA_VERSION,
    SupportState,
};

const FAKE_ID: &str = "ccd";
const FAKE_VERSION: &str = "0.1.0";

fn fake_ccd_bin() -> PathBuf {
    PathBuf::from(env!("CARGO_BIN_EXE_lifeloop-fake-ccd-client"))
}

fn manifest() -> AdapterManifest {
    let mut placement = BTreeMap::new();
    placement.insert(
        ManifestPlacementClass::PreFrameTrailing,
        ManifestPlacementSupport {
            support: SupportState::Native,
            max_bytes: None,
        },
    );
    AdapterManifest {
        contract_version: SCHEMA_VERSION.to_string(),
        adapter_id: FAKE_ID.into(),
        adapter_version: FAKE_VERSION.into(),
        display_name: "Fake CCD".into(),
        role: AdapterRole::PrimaryWorker,
        integration_modes: vec![IntegrationMode::NativeHook],
        lifecycle_events: BTreeMap::new(),
        placement,
        context_pressure: ManifestContextPressure {
            support: SupportState::Native,
            evidence: None,
        },
        receipts: ManifestReceipts {
            native: false,
            lifeloop_synthesized: true,
            receipt_ledger: SupportState::Unavailable,
        },
        session_identity: None,
        session_rename: None,
        renewal: None,
        approval_surface: None,
        failure_modes: Vec::new(),
        telemetry_sources: Vec::new(),
        known_degradations: Vec::new(),
    }
}

struct Fixture(AdapterManifest);

impl AdapterRegistry for Fixture {
    fn resolve(&self, id: &str, version: &str) -> AdapterResolution {
        if id != self.0.adapter_id {
            return AdapterResolution::UnknownId;
        }
        if version != self.0.adapter_version {
            return AdapterResolution::VersionMismatch {
                registered_version: self.0.adapter_version.clone(),
            };
        }
        AdapterResolution::Found(RegisteredAdapter {
            manifest: self.0.clone(),
            conformance: ConformanceLevel::PreConformance,
        })
    }
}

fn frame_request() -> CallbackRequest {
    CallbackRequest {
        schema_version: SCHEMA_VERSION.to_string(),
        event: LifecycleEventKind::FrameOpening,
        event_id: "evt-e2e-1".into(),
        adapter_id: FAKE_ID.into(),
        adapter_version: FAKE_VERSION.into(),
        integration_mode: IntegrationMode::NativeHook,
        invocation_id: "inv-e2e-1".into(),
        harness_session_id: Some("sess-e2e-1".into()),
        harness_run_id: Some("run-e2e-1".into()),
        harness_task_id: None,
        frame_context: Some(FrameContext::top_level("frm-e2e-1")),
        capability_snapshot_ref: None,
        payload_refs: vec![PayloadRef {
            payload_id: "pay-ref-1".into(),
            payload_kind: "instruction_frame".into(),
            content_digest: None,
            byte_size: Some(11),
        }],
        sequence: None,
        idempotency_key: Some("idem-e2e-1".into()),
        metadata: serde_json::Map::new(),
    }
}

fn build_plan(req: &CallbackRequest) -> RoutingPlan {
    let fx = Fixture(manifest());
    route(req, &fx).expect("plan builds")
}

fn invoker_with(behavior: &str, timeout: Duration) -> SubprocessCallbackInvoker {
    // Behavior is carried by a CLI arg so each spawn is independent
    // of process-global env state — cargo runs integration tests in
    // parallel and env vars would race.
    let cfg = SubprocessInvokerConfig::new(fake_ccd_bin(), timeout).arg(behavior);
    SubprocessCallbackInvoker::new(cfg)
}

// ---------------------------------------------------------------------------
// Happy paths
// ---------------------------------------------------------------------------

#[test]
fn frame_opening_returns_delivered_with_payload() {
    // Lifeloop-owned: the request shape on stdin matches the validated plan.
    // CCD-owned: the specific payload_kind/body returned.
    let invoker = invoker_with("ok", Duration::from_secs(5));
    let req = frame_request();
    let plan = build_plan(&req);
    let resp = invoker.invoke(&plan, &[]).expect("subprocess returns ok");
    assert_eq!(resp.status, ReceiptStatus::Delivered); // Lifeloop-owned (valid status)
    assert_eq!(resp.client_payloads.len(), 1); // CCD-owned (count is up to client)
    let p = &resp.client_payloads[0];
    assert_eq!(p.client_id, "ccd"); // CCD-owned
    assert_eq!(p.payload_kind, "ccd.instruction_frame"); // CCD-owned
    assert_eq!(p.idempotency_key.as_deref(), Some("idem-e2e-1")); // Lifeloop-owned (mirrored from request)
}

#[test]
fn frame_opening_delivers_payload_envelope_to_subprocess() {
    // Lifeloop-owned (issue #22): payload envelopes pass through the
    // subprocess boundary verbatim. The fake client echoes the first
    // delivered payload's body and kind into its response so we can
    // prove the bytes travelled, not that the request envelope merely
    // contained matching payload_refs.
    let invoker = invoker_with("ok", Duration::from_secs(5));
    let req = frame_request();
    let plan = build_plan(&req);

    let body = "round-trip-body-from-e2e";
    let payload = PayloadEnvelope {
        schema_version: SCHEMA_VERSION.to_string(),
        payload_id: "pay-e2e-payload-1".into(),
        client_id: "ccd".into(),
        payload_kind: "ccd.test_frame".into(),
        format: "client-defined".into(),
        content_encoding: "utf8".into(),
        body: Some(body.into()),
        body_ref: None,
        byte_size: body.len() as u64,
        content_digest: None,
        acceptable_placements: vec![AcceptablePlacement {
            placement: PlacementClass::PrePromptFrame,
            requirement: RequirementLevel::Preferred,
        }],
        idempotency_key: None,
        expires_at_epoch_s: None,
        redaction: None,
        metadata: serde_json::Map::new(),
    };

    let resp = invoker
        .invoke(&plan, std::slice::from_ref(&payload))
        .expect("subprocess returns ok");
    assert_eq!(resp.status, ReceiptStatus::Delivered);
    assert_eq!(resp.client_payloads.len(), 1);
    let echoed = &resp.client_payloads[0];
    // The fake client mirrors the delivered body and kind back —
    // proving the dispatch envelope's payloads slot reached the
    // child process and was deserializable on the client side.
    assert_eq!(echoed.payload_kind, "ccd.test_frame");
    assert_eq!(echoed.body.as_deref(), Some(body));
}

#[test]
fn session_started_returns_delivered_with_no_payload() {
    let invoker = invoker_with("ok", Duration::from_secs(5));
    let mut req = frame_request();
    req.event = LifecycleEventKind::SessionStarted;
    req.frame_context = None; // not frame-scoped
    let plan = build_plan(&req);
    let resp = invoker.invoke(&plan, &[]).expect("subprocess returns ok");
    assert_eq!(resp.status, ReceiptStatus::Delivered);
    assert!(resp.client_payloads.is_empty()); // CCD-owned
}

// ---------------------------------------------------------------------------
// Failure paths
// ---------------------------------------------------------------------------

#[test]
fn malformed_stdout_maps_to_invalid_request() {
    // Lifeloop-owned: a malformed wire response is invalid_request, not
    // transport_error — the bytes were delivered, they just don't speak
    // the contract.
    let invoker = invoker_with("malformed", Duration::from_secs(5));
    let plan = build_plan(&frame_request());
    let err = invoker.invoke(&plan, &[]).unwrap_err();
    assert!(
        matches!(err, SubprocessInvokerError::ParseResponse(_)),
        "expected ParseResponse, got {err:?}"
    );
    let fc: FailureClass = (&err).into();
    assert_eq!(fc, FailureClass::InvalidRequest);
}

#[test]
fn nonzero_exit_maps_to_transport_error() {
    // Lifeloop-owned: a non-zero child is a transport-class failure; the
    // shared FailureClass vocabulary already has TransportError, no new
    // variant needed.
    let invoker = invoker_with("nonzero", Duration::from_secs(5));
    let plan = build_plan(&frame_request());
    let err = invoker.invoke(&plan, &[]).unwrap_err();
    match &err {
        SubprocessInvokerError::NonZeroExit { code, stderr } => {
            assert_eq!(*code, Some(7));
            assert!(stderr.contains("simulated transport failure"));
        }
        other => panic!("expected NonZeroExit, got {other:?}"),
    }
    let fc: FailureClass = (&err).into();
    assert_eq!(fc, FailureClass::TransportError);
}

#[test]
fn nonzero_exit_dominates_broken_pipe_from_large_request() {
    let invoker = invoker_with("nonzero", Duration::from_secs(5));
    let plan = build_plan(&frame_request());
    let body = "x".repeat(2 * 1024 * 1024);
    let payload = PayloadEnvelope {
        schema_version: SCHEMA_VERSION.to_string(),
        payload_id: "pay-large-broken-pipe".into(),
        client_id: "ccd".into(),
        payload_kind: "ccd.large_request".into(),
        format: "client-defined".into(),
        content_encoding: "utf8".into(),
        body: Some(body.clone()),
        body_ref: None,
        byte_size: body.len() as u64,
        content_digest: None,
        acceptable_placements: vec![AcceptablePlacement {
            placement: PlacementClass::PrePromptFrame,
            requirement: RequirementLevel::Preferred,
        }],
        idempotency_key: None,
        expires_at_epoch_s: None,
        redaction: None,
        metadata: serde_json::Map::new(),
    };

    let err = invoker
        .invoke(&plan, std::slice::from_ref(&payload))
        .unwrap_err();
    match &err {
        SubprocessInvokerError::NonZeroExit { code, stderr } => {
            assert_eq!(*code, Some(7));
            assert!(stderr.contains("simulated transport failure"));
        }
        other => panic!("expected NonZeroExit to dominate BrokenPipe, got {other:?}"),
    }
    let fc: FailureClass = (&err).into();
    assert_eq!(fc, FailureClass::TransportError);
}

#[test]
fn zero_exit_broken_pipe_surfaces_write_request() {
    let invoker = invoker_with("exit_zero_no_read", Duration::from_secs(5));
    let plan = build_plan(&frame_request());
    let body = "x".repeat(2 * 1024 * 1024);
    let payload = PayloadEnvelope {
        schema_version: SCHEMA_VERSION.to_string(),
        payload_id: "pay-large-zero-broken-pipe".into(),
        client_id: "ccd".into(),
        payload_kind: "ccd.large_request".into(),
        format: "client-defined".into(),
        content_encoding: "utf8".into(),
        body: Some(body.clone()),
        body_ref: None,
        byte_size: body.len() as u64,
        content_digest: None,
        acceptable_placements: vec![AcceptablePlacement {
            placement: PlacementClass::PrePromptFrame,
            requirement: RequirementLevel::Preferred,
        }],
        idempotency_key: None,
        expires_at_epoch_s: None,
        redaction: None,
        metadata: serde_json::Map::new(),
    };

    let err = invoker
        .invoke(&plan, std::slice::from_ref(&payload))
        .unwrap_err();
    match &err {
        SubprocessInvokerError::WriteRequest(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {}
        other => panic!("expected WriteRequest(BrokenPipe), got {other:?}"),
    }
    let fc: FailureClass = (&err).into();
    assert_eq!(fc, FailureClass::TransportError);
}

#[test]
fn hung_subprocess_is_killed_after_timeout() {
    // Lifeloop-owned: a hang past the deadline is mapped to Timeout and
    // the child is killed (the test process must not leak it).
    let invoker = invoker_with("hang", Duration::from_millis(150));
    let plan = build_plan(&frame_request());
    let err = invoker.invoke(&plan, &[]).unwrap_err();
    assert!(
        matches!(err, SubprocessInvokerError::Timeout),
        "expected Timeout, got {err:?}"
    );
    let fc: FailureClass = (&err).into();
    assert_eq!(fc, FailureClass::Timeout);
}

#[test]
fn oversized_stdout_is_rejected_without_parsing() {
    let invoker = invoker_with("huge_stdout", Duration::from_secs(5));
    let plan = build_plan(&frame_request());
    let err = invoker.invoke(&plan, &[]).unwrap_err();
    assert!(
        matches!(err, SubprocessInvokerError::ReadResponse(_)),
        "expected ReadResponse for oversized stdout, got {err:?}"
    );
    let fc: FailureClass = (&err).into();
    assert_eq!(fc, FailureClass::TransportError);
}

#[cfg(unix)]
#[test]
fn inherited_stdout_pipe_times_out_and_is_cleaned_up() {
    let invoker = invoker_with("hold_stdout_open", Duration::from_millis(150));
    let plan = build_plan(&frame_request());
    let err = invoker.invoke(&plan, &[]).unwrap_err();
    assert!(
        matches!(err, SubprocessInvokerError::Timeout),
        "expected Timeout for inherited stdout pipe, got {err:?}"
    );
}

#[test]
fn receipt_emitted_is_rejected_before_spawn() {
    // Lifeloop-owned: receipt.emitted is a notification event; the
    // invoker must refuse it without spawning a child. We point the
    // config at a path that does NOT exist — if the invoker tried to
    // spawn we would see a Spawn error; instead we should see
    // ReceiptEmittedRejected.
    let cfg = SubprocessInvokerConfig::new(
        PathBuf::from("/definitely/does/not/exist/lifeloop-fake-ccd-client-missing"),
        Duration::from_secs(5),
    );
    let invoker = SubprocessCallbackInvoker::new(cfg);

    let mut req = frame_request();
    req.event = LifecycleEventKind::ReceiptEmitted;
    req.frame_context = None;
    req.idempotency_key = None; // notification events do not carry idempotency keys
    let plan = build_plan(&req);

    let err = invoker.invoke(&plan, &[]).unwrap_err();
    assert!(
        matches!(err, SubprocessInvokerError::ReceiptEmittedRejected(_)),
        "expected ReceiptEmittedRejected (pre-spawn guard), got {err:?}"
    );
    let fc: FailureClass = (&err).into();
    assert_eq!(fc, FailureClass::InvalidRequest);
}