cellos-supervisor 0.6.0-pre

//! C03 (S17/S18, ADR-0029): startup spool→broker reconcile.
//!
//! On launch under the airgapped profile, forward any rows the durable spool
//! accumulated while disconnected into the broker — idempotent via the durable
//! forward-cursor plus the server-side `duplicate_window` (S16) keyed by
//! `Nats-Msg-Id = row_hash` — then sign and emit the S18 reconcile receipt.
//!
//! The pass is **off the cell-admission critical path**: a broker that is still
//! unreachable logs a warning and startup proceeds (the spool stays durable and
//! the next launch retries). Only a corrupt durable chain is fatal, matching the
//! airgapped fail-closed posture.
//!
//! The broker publish is abstracted behind [`cellos_sink_spool::RowPublisher`]
//! so the orchestration is exercised with a mock — the live-broker behaviour is
//! proven by the S17 `forwarder_jetstream` end-to-end test.

use crate::event_signing::wrap_signed_envelope;
use async_nats::jetstream;
use async_trait::async_trait;
use cellos_core::ports::EventSink;
use cellos_core::{sign_event_with, CellosError, ExecutionCellDocument, Signer};
use cellos_sink_spool::{ChainForwarder, ForwardOutcome, RowPublisher};
use std::path::Path;
use std::sync::Arc;

/// Production [`RowPublisher`] over a JetStream context. Publishes each chain row
/// keyed by `Nats-Msg-Id = row_hash` so the server-side `duplicate_window` (S16)
/// dedups a cursor-reset replay. Uses its own connection — reconcile is a
/// one-shot startup operation, not the per-event hot path.
pub struct NatsRowPublisher {
    ctx: jetstream::Context,
    subject: String,
}

impl NatsRowPublisher {
    /// Connect to `nats_url` and target `subject`. Mirrors the broker-sink TLS
    /// surface (`NATS_CA_FILE` root for `tls://` URLs).
    ///
    /// # Errors
    ///
    /// Returns [`CellosError`] when the NATS connection cannot be established —
    /// the caller treats this as broker-unreachable and soft-fails reconcile.
    pub async fn connect(
        nats_url: &str,
        subject: impl Into<String>,
        root_ca_pem_file: Option<&Path>,
    ) -> Result<Self, CellosError> {
        let mut opts = async_nats::ConnectOptions::new();
        if let Some(p) = root_ca_pem_file {
            opts = opts.add_root_certificates(p.to_path_buf());
        }
        let conn = opts
            .connect(nats_url)
            .await
            .map_err(|e| CellosError::EventSink(format!("reconcile nats connect: {e}")))?;
        Ok(Self {
            ctx: jetstream::new(conn),
            subject: subject.into(),
        })
    }
}

#[async_trait]
impl RowPublisher for NatsRowPublisher {
    async fn publish_row(&self, msg_id: &str, payload: &[u8]) -> Result<(), CellosError> {
        let mut headers = async_nats::HeaderMap::new();
        headers.insert("Nats-Msg-Id", msg_id);
        self.ctx
            .publish_with_headers(self.subject.clone(), headers, payload.to_vec().into())
            .await
            .map_err(|e| CellosError::EventSink(format!("reconcile publish: {e}")))?
            .await
            .map_err(|e| CellosError::EventSink(format!("reconcile ack: {e}")))?;
        Ok(())
    }
    async fn flush(&self) -> Result<(), CellosError> {
        Ok(())
    }
}

/// `true` when a forward error indicates the durable chain itself is corrupt
/// (fatal under airgapped) rather than a transient broker/transport failure
/// (soft-fail). The forwarder verifies the BLAKE3 chain before publishing any
/// row, so a `chain verify` failure means on-disk corruption, not a dead broker.
fn is_chain_corruption(e: &CellosError) -> bool {
    e.to_string().contains("chain verify")
}

/// Orchestrate one startup reconcile pass for the spool at `spool_path`.
///
/// Forwards rows accumulated while disconnected through `publisher`, then signs
/// and emits the S18 reconcile receipt through `receipt_sink`. Returns
/// `Ok(Some(outcome))` on a completed pass, `Ok(None)` when the broker is
/// unreachable (soft-fail — startup continues, the spool stays durable), and
/// `Err` only when the durable chain is corrupt (airgapped fail-closed).
///
/// # Errors
///
/// Returns an error only on durable-chain corruption or receipt sign/emit
/// failure — never on a transient broker outage.
pub async fn reconcile_spool_on_startup(
    spool_path: &Path,
    publisher: &dyn RowPublisher,
    signer: &dyn Signer,
    receipt_sink: &dyn EventSink,
    chain_id: &str,
) -> anyhow::Result<Option<ForwardOutcome>> {
    let forwarder = ChainForwarder::open(spool_path)
        .map_err(|e| anyhow::anyhow!("reconcile: open spool {}: {e}", spool_path.display()))?;

    let outcome = match forwarder.forward(publisher).await {
        Ok(outcome) => outcome,
        Err(e) if is_chain_corruption(&e) => {
            return Err(anyhow::anyhow!(
                "reconcile: durable spool chain is corrupt — refusing to start \
                 (airgapped fail-closed): {e}"
            ));
        }
        Err(e) => {
            tracing::warn!(
                target: "cellos.supervisor.reconcile",
                error = %e,
                "spool reconcile soft-failed (broker unreachable) — admission continues, \
                 spool stays durable, next launch retries"
            );
            return Ok(None);
        }
    };

    // Build, sign, and emit the reconcile receipt. The build re-verifies the
    // chain, so a tamper between forward and receipt is still caught.
    let event = forwarder
        .reconcile_event(&outcome, chain_id)
        .map_err(|e| anyhow::anyhow!("reconcile: build receipt: {e}"))?;
    let envelope = sign_event_with(signer, &event)
        .map_err(|e| anyhow::anyhow!("reconcile: sign receipt: {e}"))?;
    let data = serde_json::to_value(&envelope)
        .map_err(|e| anyhow::anyhow!("reconcile: serialize receipt envelope: {e}"))?;
    let wrapper = wrap_signed_envelope(&event, data);
    receipt_sink
        .emit(&wrapper)
        .await
        .map_err(|e| anyhow::anyhow!("reconcile: emit receipt: {e}"))?;

    tracing::info!(
        target: "cellos.supervisor.reconcile",
        delivered = outcome.delivered,
        from_seq = ?outcome.from_seq,
        to_seq = ?outcome.to_seq,
        "spool reconciled into broker; signed reconcile receipt emitted"
    );
    Ok(Some(outcome))
}

/// C03 production entry: build the broker publisher + signer from the
/// environment and run one startup reconcile pass, emitting the receipt through
/// the supervisor's already-composed primary sink (`receipt_sink`) so there is a
/// single writer to the durable spool.
///
/// Every precondition that is merely "not ready" (no spool dir, no Ed25519
/// signing key, broker unreachable) is a soft skip — startup proceeds. Only a
/// corrupt durable chain propagates an error (airgapped fail-closed).
///
/// # Errors
///
/// Returns an error only on durable-chain corruption or receipt sign/emit
/// failure surfaced by [`reconcile_spool_on_startup`].
pub async fn run_startup_reconcile(
    doc: &ExecutionCellDocument,
    run_id: &str,
    receipt_sink: &Arc<dyn EventSink>,
) -> anyhow::Result<()> {
    let Some(spool_dir) = std::env::var("CELLOS_SPOOL_DIR")
        .ok()
        .map(|v| v.trim().to_string())
        .filter(|v| !v.is_empty())
    else {
        // Airgapped requires CELLOS_SPOOL_DIR; the profile gate enforces it.
        return Ok(());
    };
    let spool_path = Path::new(&spool_dir).join("audit-trail.jsonl");

    let Some(signer) = crate::event_signing::ed25519_signer_from_env() else {
        tracing::warn!(
            target: "cellos.supervisor.reconcile",
            "startup reconcile skipped: no Ed25519 event-signing key configured"
        );
        return Ok(());
    };

    let nats_url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".into());
    let subject = crate::spec_input::resolve_event_subject(
        &doc.spec.id,
        run_id,
        doc.spec
            .correlation
            .as_ref()
            .and_then(|c| c.tenant_id.as_deref()),
    );
    let nats_ca = std::env::var("NATS_CA_FILE")
        .ok()
        .filter(|s| !s.trim().is_empty())
        .map(std::path::PathBuf::from);

    let publisher = match NatsRowPublisher::connect(&nats_url, subject, nats_ca.as_deref()).await {
        Ok(p) => p,
        Err(e) => {
            tracing::warn!(
                target: "cellos.supervisor.reconcile",
                error = %e,
                "startup reconcile skipped: broker unreachable (spool stays durable, next launch retries)"
            );
            return Ok(());
        }
    };

    reconcile_spool_on_startup(
        &spool_path,
        &publisher,
        &signer,
        receipt_sink.as_ref(),
        run_id,
    )
    .await
    .map(|_| ())
}

#[cfg(test)]
mod tests {
    use super::*;
    use cellos_core::{
        verify_signed_event_envelope, CloudEventV1, SignedEventEnvelopeV1, SoftwareSigner,
        TrustAnchorPublicKey, AUDIT_RECONCILED_TYPE,
    };
    use cellos_sink_jsonl::AuditChain;
    use std::collections::HashMap;
    use std::path::PathBuf;
    use std::sync::Mutex;

    /// Records the msg_ids it was asked to publish; flush is a no-op.
    struct MockPublisher {
        published: Mutex<Vec<String>>,
    }
    impl MockPublisher {
        fn new() -> Self {
            Self {
                published: Mutex::new(Vec::new()),
            }
        }
        fn count(&self) -> usize {
            self.published.lock().unwrap().len()
        }
    }
    #[async_trait]
    impl RowPublisher for MockPublisher {
        async fn publish_row(&self, msg_id: &str, _payload: &[u8]) -> Result<(), CellosError> {
            self.published.lock().unwrap().push(msg_id.to_string());
            Ok(())
        }
        async fn flush(&self) -> Result<(), CellosError> {
            Ok(())
        }
    }

    /// Simulates a broker that is still down: every publish errors with a
    /// non-`chain verify` message (so it classifies as a soft transport failure).
    struct DeadBrokerPublisher;
    #[async_trait]
    impl RowPublisher for DeadBrokerPublisher {
        async fn publish_row(&self, _msg_id: &str, _payload: &[u8]) -> Result<(), CellosError> {
            Err(CellosError::EventSink(
                "reconcile publish: broker unreachable".into(),
            ))
        }
        async fn flush(&self) -> Result<(), CellosError> {
            Ok(())
        }
    }

    /// Captures every emitted CloudEvent for assertion.
    struct CaptureSink(Mutex<Vec<CloudEventV1>>);
    #[async_trait]
    impl EventSink for CaptureSink {
        async fn emit(&self, event: &CloudEventV1) -> Result<(), CellosError> {
            self.0.lock().unwrap().push(event.clone());
            Ok(())
        }
    }

    fn signer() -> SoftwareSigner {
        SoftwareSigner::from_seed("reconcile-kid", [9u8; 32]).unwrap()
    }

    fn verify_keys(s: &SoftwareSigner) -> HashMap<String, TrustAnchorPublicKey> {
        let mut keys = HashMap::new();
        keys.insert("reconcile-kid".to_string(), s.verifying_key());
        keys
    }

    fn seed(dir: &Path, n: u64) -> PathBuf {
        let p = dir.join("spool");
        let c = AuditChain::open(&p).unwrap();
        for i in 0..n {
            c.append(
                "dev.cellos.events.cell.lifecycle.v1.started",
                serde_json::json!({ "i": i }),
            )
            .unwrap();
        }
        p
    }

    /// Unwrap the signed-envelope transport wrapper and verify the inner receipt.
    fn verified_receipt(wrapper: &CloudEventV1, s: &SoftwareSigner) -> serde_json::Value {
        let env: SignedEventEnvelopeV1 =
            serde_json::from_value(wrapper.data.clone().unwrap()).expect("unwrap envelope");
        let event = verify_signed_event_envelope(&env, &verify_keys(s), &HashMap::new())
            .expect("reconcile receipt verifies offline");
        assert_eq!(event.ty, AUDIT_RECONCILED_TYPE);
        event.data.clone().unwrap()
    }

    #[tokio::test]
    async fn reconcile_emits_signed_receipt_and_is_idempotent() {
        let dir = tempfile::tempdir().unwrap();
        let spool = seed(dir.path(), 100);
        let pubr = MockPublisher::new();
        let sink = CaptureSink(Mutex::new(Vec::new()));
        let s = signer();

        let out = reconcile_spool_on_startup(&spool, &pubr, &s, &sink, "chain-x")
            .await
            .unwrap()
            .expect("a pass with rows returns an outcome");
        assert_eq!(out.delivered, 100);
        assert_eq!(pubr.count(), 100, "all 100 rows published");

        let events = sink.0.lock().unwrap().clone();
        assert_eq!(events.len(), 1, "exactly one reconcile receipt emitted");
        let data = verified_receipt(&events[0], &s);
        assert_eq!(data["deliveredCount"], 100);
        assert_eq!(data["fromSeq"], 0);
        assert_eq!(data["toSeq"], 99);

        // Re-run: cursor at head -> 0 delivered, a positive "already reconciled".
        let out2 = reconcile_spool_on_startup(&spool, &pubr, &s, &sink, "chain-x")
            .await
            .unwrap()
            .expect("idempotent pass still returns an outcome");
        assert_eq!(out2.delivered, 0);
        assert_eq!(pubr.count(), 100, "idempotent: nothing re-published");
        let events = sink.0.lock().unwrap().clone();
        assert_eq!(events.len(), 2, "0-delivered receipt still emitted");
        assert_eq!(verified_receipt(&events[1], &s)["deliveredCount"], 0);
    }

    #[tokio::test]
    async fn reconcile_soft_fails_when_broker_unreachable() {
        let dir = tempfile::tempdir().unwrap();
        let spool = seed(dir.path(), 10);
        let sink = CaptureSink(Mutex::new(Vec::new()));
        let s = signer();

        let out = reconcile_spool_on_startup(&spool, &DeadBrokerPublisher, &s, &sink, "chain-x")
            .await
            .expect("broker-unreachable must NOT propagate an error (off admission path)");
        assert!(out.is_none(), "soft-fail returns Ok(None)");
        assert!(
            sink.0.lock().unwrap().is_empty(),
            "no receipt emitted when nothing was reconciled"
        );
    }

    #[tokio::test]
    async fn reconcile_hard_fails_on_corrupt_chain() {
        let dir = tempfile::tempdir().unwrap();
        let spool = seed(dir.path(), 8);
        // Corrupt the durable trail so the BLAKE3 chain no longer verifies.
        let trail = AuditChain::open(&spool).unwrap().jsonl_path().to_path_buf();
        let mut bytes = std::fs::read(&trail).unwrap();
        let mid = bytes.len() / 2;
        bytes[mid] ^= 0x01;
        std::fs::write(&trail, &bytes).unwrap();

        let pubr = MockPublisher::new();
        let sink = CaptureSink(Mutex::new(Vec::new()));
        let s = signer();
        let err = reconcile_spool_on_startup(&spool, &pubr, &s, &sink, "chain-x")
            .await
            .expect_err("corrupt durable chain must be fatal under airgapped");
        assert!(
            err.to_string().contains("corrupt"),
            "error must surface chain corruption; got: {err}"
        );
        assert_eq!(pubr.count(), 0, "nothing published from a corrupt chain");
    }

    /// Pin the fatal-vs-soft classifier directly: a chain-verify failure is fatal
    /// (hard-fail startup), a transport failure is soft. If forwarder.rs's error
    /// text is ever reworded away from "chain verify", this trips.
    #[test]
    fn is_chain_corruption_classifies_fatal_vs_soft() {
        assert!(
            is_chain_corruption(&CellosError::EventSink(
                "forwarder: chain verify: blake3 mismatch at row 3".into()
            )),
            "a chain-verify failure must classify as fatal corruption"
        );
        assert!(
            !is_chain_corruption(&CellosError::EventSink(
                "reconcile publish: broker unreachable".into()
            )),
            "a transport failure must classify as soft (non-fatal)"
        );
    }

    /// An empty spool reconciles to an honest zero-delivered receipt (nothing
    /// published, deliveredCount=0) rather than erroring or claiming delivery.
    #[tokio::test]
    async fn reconcile_empty_spool_emits_honest_zero_receipt() {
        let dir = tempfile::tempdir().unwrap();
        let spool = seed(dir.path(), 0);
        let pubr = MockPublisher::new();
        let sink = CaptureSink(Mutex::new(Vec::new()));
        let s = signer();

        let out = reconcile_spool_on_startup(&spool, &pubr, &s, &sink, "chain-x")
            .await
            .unwrap()
            .expect("empty spool still returns an outcome");
        assert_eq!(out.delivered, 0);
        assert_eq!(pubr.count(), 0, "nothing published from an empty spool");
        let events = sink.0.lock().unwrap().clone();
        assert_eq!(events.len(), 1, "one honest zero-delivered receipt");
        assert_eq!(verified_receipt(&events[0], &s)["deliveredCount"], 0);
    }
}