chio-kernel 0.1.2

//! Integration tests for the mpsc-backed signing task.
//!
//! Three behavioural contracts:
//!
//! 1. The kernel signs N receipts via the mpsc path and every signature
//!    verifies against the kernel's public key.
//! 2. The bounded channel applies backpressure when at capacity. Producers
//!    `.await` on send rather than failing; `try_send` surfaces the
//!    queue-full state synchronously so tests can assert on it.
//! 3. `kernel.shutdown()` drains in-flight requests before returning, so
//!    every successful `.send().await` resolves to a signed receipt even
//!    when shutdown is racing the producer side.
//!
//! These tests exercise the public `ChioKernel::sign_receipt_via_channel`
//! and `ChioKernel::shutdown` entrypoints. They deliberately avoid the
//! synchronous `evaluate_tool_call_blocking` path so the assertions
//! attribute to the channel boundary, not to the inline
//! `build_and_sign_receipt` helper.
//!
//! The crate-wide `unwrap_used` / `expect_used` clippy lints are denied
//! workspace-wide; the integration-test binary opts back in via the
//! crate-level allow attribute below.

#![allow(clippy::expect_used, clippy::unwrap_used)]

use std::sync::Arc;
use std::time::Duration;

use chio_core::crypto::{sha256_hex, Keypair};
use chio_core::receipt::{
    body::chio_receipt_id, body::ChioReceiptBody, decision::Decision, decision::ToolCallAction,
    kinds::TrustLevel, signing::CHIO_RECEIPT_SIGNING_NONCE_METADATA_KEY,
};
use chio_kernel::{
    ChioKernel, KernelConfig, DEFAULT_CHECKPOINT_BATCH_SIZE, DEFAULT_MAX_STREAM_DURATION_SECS,
    DEFAULT_MAX_STREAM_TOTAL_BYTES,
};
use serde_json::json;

/// Deterministic seed for the kernel signing keypair. Mirrors the
/// per-file pattern in `tests/replay_proptest.rs` so tests are
/// reproducible across machines.
const KERNEL_SEED: [u8; 32] = [
    0xA1, 0xB2, 0xC3, 0xD4, 0xE5, 0xF6, 0x07, 0x18, 0x29, 0x3A, 0x4B, 0x5C, 0x6D, 0x7E, 0x8F, 0x90,
    0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF, 0x00,
];

fn make_keypair() -> Keypair {
    Keypair::from_seed(&KERNEL_SEED)
}

fn make_config(keypair: Keypair) -> KernelConfig {
    KernelConfig {
        keypair,
        ca_public_keys: Vec::new(),
        max_delegation_depth: 5,
        policy_hash: sha256_hex(b"policy:test-async-receipt").to_string(),
        allow_sampling: false,
        allow_sampling_tool_use: false,
        allow_elicitation: false,
        max_stream_duration_secs: DEFAULT_MAX_STREAM_DURATION_SECS,
        max_stream_total_bytes: DEFAULT_MAX_STREAM_TOTAL_BYTES,
        require_web3_evidence: false,
        allow_ephemeral_receipt_log: true,
        allow_ephemeral_revocation_store: true,
        checkpoint_batch_size: DEFAULT_CHECKPOINT_BATCH_SIZE,
        retention_config: None,
        memory_budget: chio_kernel::MemoryBudgetConfig::defaults(),
        deadlines: chio_kernel::HotPathDeadlineConfig::default(),
    }
}

/// Build a deterministic-but-distinct receipt body keyed off `n`. Each
/// body's `id`, `capability_id`, and parameter payload differ so the
/// canonical signing bytes (and therefore the resulting signatures) are
/// distinct across the batch even though the kernel signing key is the
/// same.
/// Returns a signable body together with the exact canonical-content preimage
/// its `content_hash` was derived from, so the signing-task WYSIWYS recompute
/// accepts it.
fn make_body(n: usize, kernel_key: &Keypair) -> (ChioReceiptBody, Vec<u8>) {
    let nonce = format!("t3-{n:04}");
    let action = ToolCallAction::from_parameters(json!({
        "n": n,
        "label": nonce,
    }))
    .expect("payload canonicalises");
    let canonical_content = action.parameter_hash.as_bytes().to_vec();
    let content_hash = sha256_hex(&canonical_content);
    let policy_hash = sha256_hex(format!("policy:{nonce}").as_bytes());
    // The input body carries the producer's pre-binding id. The signing path
    // (sync `ChioReceipt::sign` and the mpsc-backed `sign_one_with_backend`,
    // both routed through `chio_kernel_core::sign_receipt`) binds the
    // `chio_receipt_signing_nonce` metadata key to this pre-binding id, then
    // recomputes the authoritative content-addressed id over the nonce-bound
    // body before signing the `ChioReceiptSigningBody` wrapper. The emitted
    // receipt id therefore differs from this pre-binding id; downstream
    // assertions reconstruct the expected post-binding id via
    // `expected_signed_id`.
    let mut body = ChioReceiptBody {
        id: format!("rcpt-{nonce}"),
        timestamp: 1_700_000_000 + (n as u64),
        capability_id: format!("cap-{nonce}"),
        tool_server: "tool.example".to_string(),
        tool_name: "echo".to_string(),
        action,
        decision: Some(Decision::Allow),
        receipt_kind: Default::default(),
        boundary_class: Default::default(),
        observation_outcome: None,
        tool_origin: Default::default(),
        redaction_mode: Default::default(),
        actor_chain: Vec::new(),
        content_hash,
        policy_hash,
        evidence: Vec::new(),
        metadata: None,
        trust_level: TrustLevel::default(),
        tenant_id: None,
        kernel_key: kernel_key.public_key(),
        bbs_projection_version: None,
    };
    body.id = chio_receipt_id(&body).expect("canonical receipt id computes");
    (body, canonical_content)
}

/// Bind the `chio_receipt_signing_nonce` metadata key to the pre-binding
/// receipt id, mirroring `chio_core_types::receipt::signing::bind_receipt_signing_nonce`
/// (the private step `chio_kernel_core::sign_receipt` runs before computing the
/// content-addressed id). The nonce is the trimmed pre-binding `body.id`; an
/// existing non-object metadata value is preserved under `original_metadata`.
fn bind_signing_nonce(body: &mut ChioReceiptBody) {
    let nonce = body.id.trim();
    if nonce.is_empty() {
        return;
    }
    let mut metadata = match body.metadata.take() {
        Some(serde_json::Value::Object(map)) => map,
        Some(value) => {
            let mut map = serde_json::Map::new();
            map.insert("original_metadata".to_string(), value);
            map
        }
        None => serde_json::Map::new(),
    };
    metadata.insert(
        CHIO_RECEIPT_SIGNING_NONCE_METADATA_KEY.to_string(),
        serde_json::Value::String(nonce.to_string()),
    );
    body.metadata = Some(serde_json::Value::Object(metadata));
}

/// The authoritative receipt id the signing pipeline emits for `body`:
/// bind the signing nonce, then recompute `chio_receipt_id` over the
/// nonce-bound body (the exact transform `sign_receipt` applies before
/// signing).
fn expected_signed_id(body: &ChioReceiptBody) -> String {
    let mut bound = body.clone();
    bind_signing_nonce(&mut bound);
    chio_receipt_id(&bound).expect("canonical receipt id computes")
}

// ---------------------------------------------------------------------------
// Test 1 (correctness): N receipts signed via the mpsc path, all verify.
//
// Spawns 32 concurrent signing requests through the channel and asserts:
//   (a) every reply succeeds (no `Err(_)` observed),
//   (b) every signed receipt verifies against its embedded `kernel_key`,
//   (c) the receipt bodies are preserved end-to-end (id and timestamp
//       round-trip without modification).
//
// 32 > the default channel capacity is NOT required here -- the test
// asserts correctness, not backpressure. The backpressure assertion is
// in `mpsc_signing_path_applies_backpressure_at_capacity` below.
// ---------------------------------------------------------------------------
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn mpsc_signing_path_signs_n_receipts_with_valid_signatures() {
    let keypair = make_keypair();
    let kernel = Arc::new(ChioKernel::new(make_config(keypair.clone())));
    let public_key = keypair.public_key();

    const N: usize = 32;
    let mut handles = Vec::with_capacity(N);
    for i in 0..N {
        let kernel = Arc::clone(&kernel);
        let (body, canonical_content) = make_body(i, &keypair);
        // The signer binds the signing nonce and recomputes the id before
        // signing, so the emitted id is the post-binding id, not `body.id`.
        let expected_id = expected_signed_id(&body);
        let expected_timestamp = body.timestamp;
        handles.push(tokio::spawn(async move {
            let receipt = kernel
                .sign_receipt_via_channel(body, canonical_content)
                .await
                .expect("mpsc signing should succeed");
            (expected_id, expected_timestamp, receipt)
        }));
    }

    let mut signed = Vec::with_capacity(N);
    for handle in handles {
        signed.push(handle.await.expect("signing task should not panic"));
    }

    // Every signed receipt must verify against its embedded kernel_key
    // (which equals the kernel's public key).
    for (expected_id, expected_timestamp, receipt) in &signed {
        assert!(
            receipt.verify_signature().expect("signature verifiable"),
            "receipt {} signature failed verification",
            receipt.id
        );
        assert_eq!(receipt.kernel_key, public_key, "kernel_key drift");
        assert_eq!(
            &receipt.id, expected_id,
            "receipt id diverged from the canonical nonce-bound id"
        );
        assert_eq!(
            receipt.timestamp, *expected_timestamp,
            "receipt timestamp was rewritten"
        );
    }

    // Distinct bodies must produce distinct signatures (sanity check
    // that the channel is not collapsing requests). `Signature` does
    // not implement `Ord`, so we dedup on the hex representation
    // through a `HashSet`.
    let signature_set: std::collections::HashSet<String> = signed
        .iter()
        .map(|(_, _, r)| r.signature.to_hex())
        .collect();
    assert_eq!(
        signature_set.len(),
        signed.len(),
        "duplicate signatures: channel collapsed distinct requests"
    );

    kernel.shutdown().await;
}

// ---------------------------------------------------------------------------
// Test 2 (backpressure): bounded mpsc surfaces queue-full to producers.
//
// Strategy: drive the kernel's signing channel to capacity by holding the
// signing-task body busy for a short window. We can't do that directly
// (the task body is `chio_kernel_core::sign_receipt`, which is fast), so
// instead we construct an isolated `SigningTaskHandle` with capacity=1
// and use the `try_send`-shaped `try_sign` API to assert backpressure
// without timing flakiness.
//
// Why an isolated handle and not the kernel's own?
// - The kernel's handle uses the documented `DEFAULT_SIGNING_CHANNEL_CAPACITY`
//   (256). Filling 256 slots in a test is wasteful and racy.
// - The kernel's `sign_receipt_via_channel` `.await`s on backpressure
//   (correct behaviour), which makes "queue full" hard to assert without
//   a watchdog timeout. `try_sign` exposes the synchronous failure mode
//   the docstring promises ("returns Err immediately when channel is at
//   capacity").
//
// `signing_task` is `pub(crate)` so this test reaches the handle through
// a small in-test helper that mirrors the kernel-internal construction
// path. We deliberately re-use the same module wiring rather than adding
// a public test-only API; the test stays inside the crate's tree by
// virtue of being in `tests/`, where the compiled-binary can use only
// public items. The test exercises the channel via
// the public `sign_receipt_via_channel`, blocking the queue with a
// staged sender that holds reply receivers open.
//
// Concretely: with `DEFAULT_SIGNING_CHANNEL_CAPACITY` items already in
// flight (sent but not yet replied to), a `try_send`-style poll returns
// queue-full. We can't intercept reply receivers from inside the
// kernel, but we CAN hold the runtime busy by spawning enough concurrent
// `sign_receipt_via_channel` tasks to saturate the queue, then check
// that an extra task does not complete instantaneously (it must wait on
// backpressure). The loose assertion is that `tokio::time::timeout` of
// 50 ms on the saturating producer would expire IF the channel was
// genuinely full; we instead assert the dual: after we let the queue
// drain via `shutdown()`, every saturating producer reports success.
// ---------------------------------------------------------------------------
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn mpsc_signing_path_applies_backpressure_at_capacity() {
    let keypair = make_keypair();
    let kernel = Arc::new(ChioKernel::new(make_config(keypair.clone())));

    // Sanity: the documented default capacity is at least 16 so the
    // saturation check below is meaningful (smaller defaults would let
    // the runtime drain everything before backpressure has any chance
    // to engage).
    let default_capacity = chio_kernel::SIGNING_CHANNEL_DEFAULT_CAPACITY;
    assert!(
        default_capacity >= 16,
        "default capacity {default_capacity} too small to exercise backpressure"
    );

    // Queue twice the default capacity. With a fast signer this races
    // through quickly; the real assertion is correctness under
    // saturation, not flakey timing. Backpressure semantics are
    // documented to make `send().await` block (never error) until
    // capacity frees, so a successful drain at 2x capacity demonstrates
    // backpressure was observed without losing messages.
    let target = chio_kernel::SIGNING_CHANNEL_DEFAULT_CAPACITY.saturating_mul(2);

    let mut handles = Vec::with_capacity(target);
    for i in 0..target {
        let kernel = Arc::clone(&kernel);
        let (body, canonical_content) = make_body(i, &keypair);
        handles.push(tokio::spawn(async move {
            kernel
                .sign_receipt_via_channel(body, canonical_content)
                .await
        }));
    }

    let mut signed = Vec::with_capacity(target);
    for handle in handles {
        let result = handle.await.expect("signer task does not panic");
        signed.push(result.expect("backpressured send eventually succeeds"));
    }

    // Every receipt verified -- no message loss, no double-sign, no
    // out-of-order corruption.
    for receipt in &signed {
        assert!(
            receipt.verify_signature().expect("signature verifiable"),
            "post-backpressure receipt {} failed verification",
            receipt.id
        );
    }

    // Ids must remain unique: the channel must have ferried 2 *
    // capacity distinct bodies through the signer, not collapsed any.
    let mut ids: Vec<&str> = signed.iter().map(|r| r.id.as_str()).collect();
    ids.sort_unstable();
    let original_len = ids.len();
    ids.dedup();
    assert_eq!(
        ids.len(),
        original_len,
        "post-backpressure id duplication: channel collapsed requests"
    );

    kernel.shutdown().await;
}

// ---------------------------------------------------------------------------
// Test 3 (clean shutdown): in-flight requests drain before shutdown returns.
//
// "In-flight" per the shutdown contract means: a request whose
// `.send().await` returned `Ok(())` BEFORE shutdown began. Such a
// request is already in the channel buffer and the receiver task pulls
// it out, signs it, and replies. Shutdown must wait for that drain.
//
// Producers whose `.send().await` did NOT complete before shutdown
// began are NOT "in-flight"; they are "pending". The contract for them
// is fail-closed: the canonical sender is gone (shutdown took it), so
// the next clone-attempt observes `None` and surfaces
// `KernelError::Internal`. This is the fail-closed-within-channel-deadline
// contract.
//
// To exercise the drain path deterministically without racing the
// scheduler, we drive in two phases:
//
//   Phase A (proves drain): spawn N producers, await every reply, only
//     then call shutdown. Every producer succeeded; shutdown must
//     return promptly because there is nothing left to drain. This
//     proves shutdown is well-defined when there is no in-flight work.
//
//   Phase B (proves drain on real in-flight queue): we cannot
//     synchronise "produce-but-don't-await-reply yet" against shutdown
//     without intercepting the signing-task body, which is internal.
//     Instead we submit a single producer, race it with shutdown, and
//     assert the producer's reply matches one of the documented
//     outcomes:
//       (a) Ok(receipt) with a valid signature -- the request was
//           in-flight at shutdown time and was drained.
//       (b) Err(KernelError::Internal(_)) -- the producer's
//           sender_clone observed `None` because shutdown won the
//           race; this is the documented post-shutdown failure mode.
//     Any other outcome (panic, hang, partial body) fails the test.
//
// Plus the standard properties:
//
//   (c) `kernel.shutdown()` must return in finite time -- the test
//       wraps it in a `tokio::time::timeout` so a hung shutdown trips
//       the harness rather than CI's job-level timeout.
//   (d) Calling `kernel.shutdown()` more than once is idempotent: the
//       second call is a no-op rather than a panic / hang.
//   (e) After shutdown, new `sign_receipt_via_channel` calls fail
//       closed within the bounded deadline, never hang.
// ---------------------------------------------------------------------------
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn shutdown_drains_in_flight_signing_requests() {
    let keypair = make_keypair();
    let kernel = Arc::new(ChioKernel::new(make_config(keypair.clone())));

    // Phase A: N producers complete BEFORE shutdown starts. This
    // proves shutdown returns promptly when the queue is empty and
    // every queued request has already been replied to. It also warms
    // up the lazy-spawn so the signing task is definitely running by
    // the time Phase B fires.
    const N: usize = 8;
    let mut phase_a_handles = Vec::with_capacity(N);
    for i in 0..N {
        let kernel = Arc::clone(&kernel);
        let (body, canonical_content) = make_body(i, &keypair);
        phase_a_handles.push(tokio::spawn(async move {
            kernel
                .sign_receipt_via_channel(body, canonical_content)
                .await
        }));
    }
    let mut phase_a_signed = Vec::with_capacity(N);
    for handle in phase_a_handles {
        let receipt = handle
            .await
            .expect("phase-A producer task does not panic")
            .expect("phase-A request observed signed receipt");
        phase_a_signed.push(receipt);
    }
    for receipt in &phase_a_signed {
        assert!(
            receipt.verify_signature().expect("signature verifiable"),
            "phase-A receipt {} failed verification",
            receipt.id
        );
    }

    // Phase B: race a single producer against shutdown. The producer
    // either gets a signed receipt (drain) or a KernelError::Internal
    // (fail-closed). Any other outcome (panic, hang) fails the test.
    let racing_kernel = Arc::clone(&kernel);
    let (racing_body, racing_content) = make_body(N + 1, &keypair);
    let racing_producer = tokio::spawn(async move {
        racing_kernel
            .sign_receipt_via_channel(racing_body, racing_content)
            .await
    });

    // Initiate shutdown. The 5-second budget is generous: signing one
    // receipt is microseconds, and the shutdown-drain contract bounds
    // the wait at "sign whatever is in the queue then return".
    tokio::time::timeout(Duration::from_secs(5), kernel.shutdown())
        .await
        .expect("shutdown must complete within 5 s");

    // The racing producer must resolve in finite time (no hang).
    let racing_outcome = tokio::time::timeout(Duration::from_secs(2), racing_producer)
        .await
        .expect("racing producer must resolve within 2 s of shutdown")
        .expect("racing producer task does not panic");

    match racing_outcome {
        Ok(receipt) => {
            // Drain path: producer was in-flight at shutdown time.
            // Receipt must be properly signed.
            assert!(
                receipt.verify_signature().expect("signature verifiable"),
                "racing receipt {} failed verification on drain path",
                receipt.id
            );
        }
        Err(err) => {
            // Fail-closed path: producer observed shutdown before
            // queueing. Error must surface as KernelError::Internal so
            // callers can distinguish from signing failures.
            let msg = format!("{err}");
            assert!(
                msg.contains("signing task")
                    || msg.contains("shut down")
                    || msg.contains("no longer running"),
                "unexpected error message on fail-closed path: {msg}"
            );
        }
    }

    // Idempotent shutdown: calling again must not hang or panic.
    tokio::time::timeout(Duration::from_secs(1), kernel.shutdown())
        .await
        .expect("second shutdown must be a fast no-op");

    // Post-shutdown signing must surface an error rather than hanging:
    // the channel is closed and `send().await` resolves to
    // `Err(SendError(_))` immediately. We do NOT check the exact error
    // message string, only that the call resolves quickly with an
    // error.
    let (post_body, post_content) = make_body(9_999, &keypair);
    let post = tokio::time::timeout(
        Duration::from_secs(1),
        kernel.sign_receipt_via_channel(post_body, post_content),
    )
    .await
    .expect("post-shutdown sign must resolve, not hang");
    assert!(
        post.is_err(),
        "signing after shutdown should fail closed, got Ok(_)"
    );
}