//! T6.B / J-H4 — D1–D12 doctrine red-line audit (cellos-core surface).
//!
//! Each `dN_…` test pins one item from the doctrine red-line matrix in
//! `docs/doctrine-coverage.md`. Tests in this file cover only properties
//! observable from `cellos-core` itself — supervisor-side ingress hygiene
//! (D7) lives in `crates/cellos-supervisor/tests/doctrine.rs`, and the
//! end-to-end "ambient host env does not leak into the child" check (D1)
//! lives in `crates/cellos-supervisor/tests/supervisor_no_ambient_env.rs`.
//!
//! When a doctrine rule is not testable as code (D6, D8) the rationale
//! is recorded in `docs/doctrine-coverage.md`.

use std::fs;
use std::path::{Path, PathBuf};

use cellos_core::authority::{
    canonical_class_for_rule, AppliedRule, AuthorityDerivation, AuthorityInput, AuthorityInputType,
    AuthorityOutput, BindingStatus, EpistemicStatus, ImposedAuthority, ObservedAuthority,
    ProvenAuthority, ProvenAuthorityArtifact, Rule, RuleClass, ValidationError,
};
use cellos_core::ports::{CellBackend, EventSink, ExportSink, SecretBroker};
use cellos_core::{
    lifecycle_destroyed_data_v1, lifecycle_started_data_v1, AuthorityBundle, ExecutionCellSpec,
    LifecycleDestroyOutcome, Lifetime, NoopSecretBroker,
};

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Strip line- and block-comments from a Rust source string so that grep-
/// style D11 checks don't false-fire on prose like `// no env::var here`.
///
/// This is intentionally simple and **not** a Rust parser — it handles the
/// two comment forms that show up in `cellos-core` source. String literals
/// containing the literal substring `//` would be misclassified, but
/// `cellos-core` doesn't ship any such literal in non-test code (the only
/// "//" substrings are in tests' redact-URL fixtures, and tests are excluded
/// from the D11 walk below).
fn strip_rust_comments(src: &str) -> String {
    let bytes = src.as_bytes();
    let mut out = String::with_capacity(src.len());
    let mut i = 0;
    while i < bytes.len() {
        // Block comment.
        if i + 1 < bytes.len() && bytes[i] == b'/' && bytes[i + 1] == b'*' {
            let mut j = i + 2;
            while j + 1 < bytes.len() && !(bytes[j] == b'*' && bytes[j + 1] == b'/') {
                // Preserve newlines so line numbers in panic messages stay
                // aligned with the source file.
                if bytes[j] == b'\n' {
                    out.push('\n');
                }
                j += 1;
            }
            i = j + 2;
            continue;
        }
        // Line comment.
        if i + 1 < bytes.len() && bytes[i] == b'/' && bytes[i + 1] == b'/' {
            while i < bytes.len() && bytes[i] != b'\n' {
                i += 1;
            }
            continue;
        }
        out.push(bytes[i] as char);
        i += 1;
    }
    out
}

/// Walk `cellos-core/src/**/*.rs` and yield `(path, comment-stripped source)`.
///
/// Excludes inline `#[cfg(test)] mod tests` and `tests/` files implicitly —
/// this walks `src/` only, where production hygiene applies.
fn walk_core_src() -> Vec<(PathBuf, String)> {
    let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
    let mut acc = Vec::new();
    fn recur(dir: &Path, acc: &mut Vec<(PathBuf, String)>) {
        for entry in fs::read_dir(dir).expect("read core src dir") {
            let entry = entry.expect("dirent");
            let path = entry.path();
            if path.is_dir() {
                recur(&path, acc);
            } else if path.extension().and_then(|e| e.to_str()) == Some("rs") {
                let raw =
                    fs::read_to_string(&path).unwrap_or_else(|_| panic!("read {}", path.display()));
                acc.push((path, strip_rust_comments(&raw)));
            }
        }
    }
    recur(&root, &mut acc);
    acc
}

fn minimal_spec(id: &str) -> ExecutionCellSpec {
    ExecutionCellSpec {
        id: id.to_string(),
        correlation: None,
        ingress: None,
        environment: None,
        placement: None,
        policy: None,
        identity: None,
        run: None,
        authority: AuthorityBundle::default(),
        lifetime: Lifetime { ttl_seconds: 60 },
        export: None,
        telemetry: None,
    }
}

// ---------------------------------------------------------------------------
// D1 — No ambient authority by default
// ---------------------------------------------------------------------------

/// A doctrine-minimal spec carries **no** ambient grants: no egress
/// allowlist, no secret refs, no environment, no ingress. The end-to-end
/// "no ambient host env leaks into the child" check is in the supervisor
/// integration test [`supervisor_no_ambient_env`]; this test pins the
/// shape of the type that represents a doctrine-clean spec.
#[test]
fn d1_minimal_spec_has_no_ambient_egress_or_secrets() {
    let spec = minimal_spec("d1-minimal");
    assert!(
        spec.authority.egress_rules.is_none()
            || spec
                .authority
                .egress_rules
                .as_ref()
                .map(Vec::is_empty)
                .unwrap_or(true),
        "D1: minimal authority must not pre-grant egress"
    );
    assert!(
        spec.authority.secret_refs.is_none()
            || spec
                .authority
                .secret_refs
                .as_ref()
                .map(Vec::is_empty)
                .unwrap_or(true),
        "D1: minimal authority must not pre-grant secret refs"
    );
    assert!(
        spec.environment.is_none(),
        "D1: minimal spec must not carry inherited environment"
    );
    assert!(
        spec.ingress.is_none(),
        "D1: minimal spec must not carry pre-declared ingress"
    );
}

// ---------------------------------------------------------------------------
// D2 — Run identity threads through every emission
// ---------------------------------------------------------------------------

#[test]
fn d2_lifecycle_started_carries_run_id() {
    let spec = minimal_spec("d2-started");
    let run_id = "run-d2-7f3a";
    let v = lifecycle_started_data_v1(
        &spec,
        "cell-d2",
        Some(run_id),
        None,
        None,
        None,
        None,
        None,
        None,
        None,
    )
    .expect("build started payload");
    let got = v
        .as_object()
        .and_then(|m| m.get("runId"))
        .and_then(|v| v.as_str())
        .expect("D2: runId must be present in lifecycle.started payload");
    assert_eq!(got, run_id, "D2: runId must round-trip verbatim");
}

#[test]
fn d2_lifecycle_destroyed_carries_run_id() {
    let spec = minimal_spec("d2-destroyed");
    let run_id = "run-d2-9c41";
    let v = lifecycle_destroyed_data_v1(
        &spec,
        "cell-d2",
        Some(run_id),
        LifecycleDestroyOutcome::Succeeded,
        None,
        None,
        None,
        None,
    )
    .expect("build destroyed payload");
    let got = v
        .as_object()
        .and_then(|m| m.get("runId"))
        .and_then(|v| v.as_str())
        .expect("D2: runId must be present in lifecycle.destroyed payload");
    assert_eq!(got, run_id, "D2: runId must round-trip verbatim");
}

// ---------------------------------------------------------------------------
// D3 — Secret resolution is per-cell (no shared "current cell")
// ---------------------------------------------------------------------------

/// `SecretBroker::resolve` MUST take an explicit `cell_id` parameter — the
/// shared-global "current cell" anti-pattern is doctrine-forbidden. This
/// test invokes the port through a `&dyn SecretBroker` so a future
/// refactor that drops `cell_id` from `resolve` or `revoke_for_cell`
/// breaks at compile time.
#[test]
fn d3_secret_broker_resolve_is_per_cell() {
    fn require_send_sync<T: Send + Sync + ?Sized>() {}
    require_send_sync::<dyn SecretBroker + Send + Sync>();

    // Touch the per-cell signatures through a trait object so the
    // (key, cell_id, ttl_seconds) and (cell_id) parameter shapes are
    // pinned by the compiler.
    let broker: &(dyn SecretBroker + Send + Sync) = &NoopSecretBroker;
    let rt = mini_block_on();
    rt.block_on(async {
        // resolve: per-cell call site.
        let _ = broker.resolve("KEY", "cell-d3-a", 30).await;
        let _ = broker.resolve("KEY", "cell-d3-b", 30).await;
        // revoke: takes cell_id, never a global.
        broker
            .revoke_for_cell("cell-d3-a")
            .await
            .expect("noop broker revoke is infallible");
    });
}

/// Build a tiny block-on shim — doesn't pull in tokio. The doctrine tests
/// don't need real async parallelism; this driver is sufficient for D3's
/// `&dyn SecretBroker` port touch (the `NoopSecretBroker` future is
/// immediately ready).
fn mini_block_on() -> StdBlockOn {
    StdBlockOn
}

/// Tiny block-on shim — uses `std`'s thread parking to drive a single
/// future to completion.
struct StdBlockOn;
impl StdBlockOn {
    fn block_on<F: std::future::Future>(&self, fut: F) -> F::Output {
        use std::sync::Arc;
        use std::task::{Context, Poll, Wake, Waker};
        struct ParkWaker(std::thread::Thread);
        impl Wake for ParkWaker {
            fn wake(self: Arc<Self>) {
                self.0.unpark();
            }
        }
        let thread = std::thread::current();
        let waker = Waker::from(Arc::new(ParkWaker(thread)));
        let mut cx = Context::from_waker(&waker);
        let mut fut = Box::pin(fut);
        loop {
            match fut.as_mut().poll(&mut cx) {
                Poll::Ready(out) => return out,
                Poll::Pending => std::thread::park(),
            }
        }
    }
}

// ---------------------------------------------------------------------------
// D4 — `EventSink` is the only emission contract
// ---------------------------------------------------------------------------

/// Asserts at compile time that `EventSink` is the **single** sink trait
/// for CloudEvent emission, and that it is `Send + Sync` so it can cross
/// task boundaries without requiring shared globals.
///
/// Together with the D11 `no_nats_or_jetstream_calls` grep, this pins
/// "core never reaches a transport directly — it hands a CloudEvent to
/// the only emission port and walks away."
#[test]
fn d4_eventsink_port_is_the_emission_contract() {
    fn require_send_sync<T: Send + Sync + ?Sized>() {}
    // `EventSink: Send + Sync` is part of the trait declaration; this
    // assertion fails to compile if the supertrait bound is dropped.
    require_send_sync::<dyn EventSink + Send + Sync>();
    // Also pin that the `emit` method is reachable through a trait
    // object — a future refactor that makes the trait non-dyn-compatible
    // (e.g. by adding a generic method) breaks this line.
    let _sink: Box<dyn EventSink + Send + Sync> = Box::new(cellos_core::ports::NoopEventSink);
}

// ---------------------------------------------------------------------------
// D5 — Teardown carries an outcome (residue is testable end-to-end)
// ---------------------------------------------------------------------------

#[test]
fn d5_lifecycle_destroyed_payload_carries_outcome() {
    let spec = minimal_spec("d5-outcome");
    for outcome in [
        LifecycleDestroyOutcome::Succeeded,
        LifecycleDestroyOutcome::Failed,
    ] {
        let v = lifecycle_destroyed_data_v1(
            &spec,
            "cell-d5",
            Some("run-d5"),
            outcome,
            None,
            None,
            None,
            None,
        )
        .expect("build destroyed payload");
        let got = v
            .as_object()
            .and_then(|m| m.get("outcome"))
            .and_then(|v| v.as_str())
            .expect("D5: outcome must be present");
        let expected = match outcome {
            LifecycleDestroyOutcome::Succeeded => "succeeded",
            LifecycleDestroyOutcome::Failed => "failed",
        };
        assert_eq!(got, expected, "D5: outcome must serialize lowercase");
    }
}

// ---------------------------------------------------------------------------
// D10 — `CellBackend` is the host seam
// ---------------------------------------------------------------------------

/// `CellBackend` is the only port that maps a spec to host isolation
/// primitives. Pin its dyn-compat and `Send + Sync` so the supervisor can
/// hold a `Box<dyn CellBackend>` across awaits without bolting on a
/// process-global host registry.
#[test]
fn d10_cellbackend_is_the_host_seam() {
    fn require_send_sync<T: Send + Sync + ?Sized>() {}
    // `CellBackend: Send + Sync` is part of the trait declaration so the
    // supervisor can hold an `Arc<dyn CellBackend>` across `await`s
    // without bolting on a process-global host registry.
    require_send_sync::<dyn CellBackend + Send + Sync>();
    // ExportSink also crosses the host seam.
    require_send_sync::<dyn ExportSink + Send + Sync>();
}

// ---------------------------------------------------------------------------
// D11 — `cellos-core` is I/O-free and globally pure
// ---------------------------------------------------------------------------

#[test]
fn d11_no_env_var_reads_in_cellos_core_src() {
    let mut offenders = Vec::new();
    for (path, src) in walk_core_src() {
        // Match `std::env::var` and bare `env::var(`. The comment-strip pass
        // above ensures prose mentions like `// std::env::var is forbidden`
        // do not false-fire.
        if src.contains("std::env::var") || src.contains("env::var(") {
            offenders.push(path);
        }
    }
    assert!(
        offenders.is_empty(),
        "D11: cellos-core/src must not read process env vars; offenders: {offenders:?}"
    );
}

#[test]
fn d11_no_nats_or_jetstream_calls_in_cellos_core_src() {
    // Tokens that would indicate a real NATS/JetStream client call.
    // Substrings that survive comment stripping but are still benign
    // (e.g. log-redaction fixtures, doc-strings already removed) are
    // guarded by the comment stripper.
    const FORBIDDEN: &[&str] = &[
        "async_nats::",
        "nats::",
        "jetstream::",
        "JetStream::",
        "::connect_nats",
    ];
    let mut offenders: Vec<(PathBuf, &'static str)> = Vec::new();
    for (path, src) in walk_core_src() {
        for needle in FORBIDDEN {
            if src.contains(needle) {
                offenders.push((path.clone(), needle));
            }
        }
    }
    assert!(
        offenders.is_empty(),
        "D11: cellos-core/src must not call NATS/JetStream; offenders: {offenders:?}"
    );
}

#[test]
fn d11_no_global_mutable_state_in_cellos_core_src() {
    // Tokens that betray a shared-global "current cell" or process-wide
    // singleton — explicitly red-lined by DOCTRINE.md.
    const FORBIDDEN: &[&str] = &[
        "static mut ",
        "lazy_static!",
        "once_cell::sync::Lazy",
        "OnceCell::new()",
        "OnceLock::new()",
    ];
    let mut offenders: Vec<(PathBuf, &'static str)> = Vec::new();
    for (path, src) in walk_core_src() {
        for needle in FORBIDDEN {
            if src.contains(needle) {
                offenders.push((path.clone(), needle));
            }
        }
    }
    assert!(
        offenders.is_empty(),
        "D11: cellos-core/src must not hold global mutable state; offenders: {offenders:?}"
    );
}

// ---------------------------------------------------------------------------
// D12 — `AuthorityDerivation` requires non-empty inputs + rules
// ---------------------------------------------------------------------------

/// Construct a derivation with a single observed-class rule firing.
/// Used as the "valid baseline" against which D12's empty-vec mutations
/// are compared.
fn baseline_observed_derivation() -> AuthorityDerivation {
    let rule = Rule::RawSniObserved;
    let class = canonical_class_for_rule(rule);
    AuthorityDerivation {
        inputs: vec![AuthorityInput {
            input_type: AuthorityInputType::Sni,
            value: "example.test".into(),
            confidence: 0.5,
            source_event_id: None,
        }],
        rules_applied: vec![AppliedRule { rule, class }],
        output: AuthorityOutput {
            tier: 1,
            confidence: 0.5,
            epistemic_status: EpistemicStatus::DirectObservation,
            binding_status: BindingStatus::NotApplicable,
        },
    }
}

#[test]
fn d12_authority_derivation_required_for_typed_authority() {
    // Baseline must validate — otherwise this test is meaningless.
    let baseline = baseline_observed_derivation();
    baseline
        .validate()
        .expect("D12 baseline derivation must validate");

    // Mutation 1 — empty inputs.
    let mut empty_inputs = baseline.clone();
    empty_inputs.inputs.clear();
    assert!(
        matches!(empty_inputs.validate(), Err(ValidationError::NoInputs)),
        "D12: empty inputs must fail with NoInputs"
    );
    // Typed-authority constructors must also reject the empty derivation.
    assert!(
        ObservedAuthority::try_new(empty_inputs.clone()).is_err(),
        "D12: ObservedAuthority must reject empty inputs"
    );
    let stub_artifact = ProvenAuthorityArtifact::JwsVerified {
        kid: "kid-d12-stub".into(),
    };
    assert!(
        ProvenAuthority::try_new(empty_inputs.clone(), stub_artifact).is_err(),
        "D12: ProvenAuthority must reject empty inputs"
    );
    assert!(
        ImposedAuthority::try_new(empty_inputs).is_err(),
        "D12: ImposedAuthority must reject empty inputs"
    );

    // Mutation 2 — empty rules_applied.
    let mut empty_rules = baseline_observed_derivation();
    empty_rules.rules_applied.clear();
    assert!(
        matches!(empty_rules.validate(), Err(ValidationError::NoRulesApplied)),
        "D12: empty rules_applied must fail with NoRulesApplied"
    );
    assert!(
        ObservedAuthority::try_new(empty_rules).is_err(),
        "D12: ObservedAuthority must reject empty rules_applied"
    );

    // Witness — `AuthorityDerivation.inputs` and `.rules_applied` are
    // declared as plain `Vec<_>`, **not** `Option<Vec<_>>`, so callers
    // cannot omit them. The mutations above prove the runtime check;
    // the type-shape declaration in `validator.rs` lines 208-210 is the
    // compile-time guarantee. RuleClass kept "live" so D9's typed
    // envelope tests do not silently lose their fixture.
    let _ = RuleClass::RawObservation;
}
