alkblobs 0.1.0

//! The ops module's shared test support and the e2e test tree (the
//! `ops-e2e-flows` capstone, the wire-level json/fetch/put handler
//! tests, the fleet renewal forms against the live harness server, and
//! the structural no-side-door audit): the harness machinery lives in
//! `harness.rs`; the unit tests ride their subject modules (`tokens.rs`,
//! `fetch.rs`).

#[cfg(feature = "postgres")]
use std::collections::BTreeSet;
use std::sync::Arc;
#[cfg(feature = "postgres")]
use std::time::Duration;

use alkcall::channels::client::ChannelClient;
use alkcall::channels::reassembly::{MpscRecvStream, MpscSendStream};
use alkcall::core::types::{BiStream, BidiStreamSource, Connection, StreamError};
use alkcall::protocol::wire::ResponseEnvelope;
use alkcall::registry::context::{CompositionAuthority, ScopedPeerEnv};
use parking_lot::Mutex;
use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

use super::harness::{
    await_call, bare_context, composer_pair, identity_named, ChannelOpenable, OpsHarness,
    OpsHarnessRegistration, TEST_TIMEOUT,
};
#[cfg(feature = "postgres")]
use crate::facade::Pin;
use crate::hashing::Digest;
use crate::keys::Key;
use crate::store::Store;
#[cfg(feature = "postgres")]
use crate::Error;

use super::*;

pub(crate) struct RecordingConverter {
    pub(crate) converted: Mutex<Vec<Key>>,
}

impl RecordingConverter {
    pub(crate) fn captured(&self) -> Vec<Key> {
        self.converted.lock().clone()
    }
}

impl PinConverter for RecordingConverter {
    fn convert(&self, key: &Key) {
        self.converted.lock().push(key.clone());
    }
}

pub(crate) fn key_of(seed: &[u8]) -> Key {
    Key::from(Digest::derive(seed, seed.len() as u64))
}

#[cfg(any(feature = "mem", feature = "sqlite"))]
pub(crate) async fn test_store() -> Arc<Store> {
    #[cfg(feature = "mem")]
    let store = Store::mem_only();
    #[cfg(not(feature = "mem"))]
    let store = {
        let path = std::path::PathBuf::from(format!(
            "/tmp/opencode/alkblobs-ops-foundation-{}-{}.db",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_nanos()
        ));
        Store::kv_only_sqlite(path, 1024, std::time::Duration::ZERO)
            .await
            .unwrap()
    };
    Arc::new(store)
}

// ----------------------------------------------------------------
// blobs/fetch (the binary data plane's consumer side): the verified
// Sub op through the channel halves.
// ----------------------------------------------------------------

#[cfg(any(feature = "mem", feature = "sqlite"))]
fn fetch_openable(ctx: Arc<OpsContext>) -> ChannelOpenable {
    let (spec, establisher, open_handler) = fetch_openables(ctx);
    ChannelOpenable::pump(spec, establisher, open_handler)
}

#[cfg(any(feature = "mem", feature = "sqlite"))]
fn put_openable(ctx: Arc<OpsContext>) -> ChannelOpenable {
    let (spec, establisher, pub_handler) = put_openables(ctx);
    ChannelOpenable::put(spec, establisher, pub_handler)
}

#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn open_fetch(
    client: &ChannelClient,
    namespace: &str,
    digest_hex: &str,
    ranges: Option<Value>,
) -> Result<(u32, Value, MpscSendStream, MpscRecvStream), alkcall::channels::client::ChannelOpenError>
{
    let mut input = json!({ "namespace": namespace, "digest": digest_hex });
    if let Some(ranges) = ranges {
        input["ranges"] = ranges;
    }
    tokio::time::timeout(
        TEST_TIMEOUT,
        client.open_channel_with_reply(FETCH_OP_ID, input, BLOBS_CHANNEL_ALPN),
    )
    .await
    .expect("the open resolves")
}

/// The whole-blob verified round-trip (the op's core criterion):
/// the seeded whole streams byte-exact over the channel and the
/// consumer-side hash-check passes against the carried digest; the
/// ack carries the store path's `{source, len}` shape.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_serves_the_verified_whole_and_the_consumer_hash_checks() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let seed: &[u8] = b"the fetched whole blob";
    let (key, _pin) = store
        .put(Some(seed.len() as u64), std::io::Cursor::new(seed))
        .await
        .expect("seed put lands");
    let digest_hex = hex::encode(key.digest().bytes());

    let (channel_id, reply, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None)
        .await
        .expect("the fetch channel establishes");
    assert!(channel_id > 0, "a data channel, not channel 0 itself");
    assert_eq!(
        reply,
        json!({ "channel_id": channel_id, "source": "store", "len": seed.len() }),
        "the store path's ack: source + the expected whole's length"
    );

    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    assert_eq!(received, seed, "byte-exact");
    assert_eq!(
        Key::from(Digest::derive(&received, received.len() as u64)),
        key,
        "the consumer's whole-blob hash-check passes"
    );
    drop(send);
}

/// The ranged form (criterion: the ADR-006 slice-digest convention,
/// out-of-band, over exactly the returned bytes): the ack carries
/// one slice digest per range in request order — read_range's
/// convention verbatim — and the served concatenation is the
/// ranges' slices; an over-read range clamps (its digest covers
/// exactly the clamped bytes).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_serves_ranges_and_the_ack_carries_the_slice_digests() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let seed: Vec<u8> = (0u8..64).collect();
    let (key, _pin) = store
        .put(Some(seed.len() as u64), std::io::Cursor::new(seed))
        .await
        .expect("seed put lands");
    let digest_hex = hex::encode(key.digest().bytes());

    let (slice0, digest0) = store
        .read_range(&key.as_bytes(), 2, 5)
        .await
        .expect("the store's ranged read");
    let (_, digest1) = store
        .read_range(&key.as_bytes(), 1000, 9999)
        .await
        .expect("the store's past-EOF ranged read");

    let (channel_id, reply, send, mut recv) = open_fetch(
        &client,
        "ns-1",
        &digest_hex,
        Some(json!([{ "offset": 2, "len": 5 }, { "offset": 1000, "len": 9999 }])),
    )
    .await
    .expect("the fetch channel establishes");
    assert_eq!(
        reply,
        json!({
            "channel_id": channel_id,
            "source": "store",
            "slice_digests": [hex::encode(digest0), hex::encode(digest1)]
        }),
        "the ack carries read_range's slice digests, one per range, in request order"
    );

    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    let mut expected = slice0;
    expected.extend(
        store
            .read_range(&key.as_bytes(), 1000, 9999)
            .await
            .expect("re-read")
            .0,
    );
    assert_eq!(received, expected, "the concatenation of the served slices");
    drop(send);
}

/// The ranged form's bounds (review-001 S-2), at their gates: an
/// over-count open is the registry's schema `INVALID_INPUT` (the
/// `maxItems` rides the context's `fetch_max_ranges` knob — the small
/// custom config here proves the knob, not a constant, is the home);
/// an over-bytes open — the ingress sums the **requested** lens, the
/// schema cannot — is the establishment `HandlerError` refusal
/// (`channel:open_failed`, the stable domain text), across one item,
/// summed items, and the checked-add overflow arm. A request at the
/// exact byte bound still establishes and byte-verifies end to end.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_over_count_rejects_at_the_schema_and_over_bytes_at_the_ingress() {
    let store = test_store().await;
    let config = OpsConfig {
        fetch_max_ranges: 3,
        fetch_max_total_bytes: 16,
        ..OpsConfig::default()
    };
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), config));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let (digest_hex, key) = seeded_entry(&store, b"bounded ranged fetch").await;

    let over_count = open_denied(
        &client,
        FETCH_OP_ID,
        json!({
            "namespace": "ns-1",
            "digest": digest_hex.clone(),
            "ranges": [
                { "offset": 0, "len": 1 },
                { "offset": 1, "len": 1 },
                { "offset": 2, "len": 1 },
                { "offset": 3, "len": 1 }
            ]
        }),
    )
    .await;
    assert_eq!(
        error_code(&over_count),
        Some("INVALID_INPUT"),
        "the schema's maxItems (the context's fetch_max_ranges) rejects the over-count \
         open before any handler: {over_count:?}"
    );

    let single_over_bytes = open_denied(
        &client,
        FETCH_OP_ID,
        json!({
            "namespace": "ns-1",
            "digest": digest_hex.clone(),
            "ranges": [{ "offset": 0, "len": 17 }]
        }),
    )
    .await;
    match single_over_bytes.result {
        Err(error) => {
            assert_eq!(error.code, "channel:open_failed", "{error:?}");
            assert!(
                error.message.contains("exceeds the total-bytes bound"),
                "the stable domain text names the bound's shape: {error:?}"
            );
        }
        Ok(_) => panic!("an over-bytes ranged open established: {single_over_bytes:?}"),
    }

    let summed_over_bytes = open_denied(
        &client,
        FETCH_OP_ID,
        json!({
            "namespace": "ns-1",
            "digest": digest_hex.clone(),
            "ranges": [
                { "offset": 0, "len": 6 },
                { "offset": 0, "len": 6 },
                { "offset": 0, "len": 6 }
            ]
        }),
    )
    .await;
    match summed_over_bytes.result {
        Err(error) => {
            assert_eq!(error.code, "channel:open_failed", "{error:?}");
            assert!(
                error.message.contains("exceeds the total-bytes bound"),
                "the schema cannot sum — the ingress does: {error:?}"
            );
        }
        Ok(_) => panic!("summed over-bytes ranged open established: {summed_over_bytes:?}"),
    }

    let overflow_arm = open_denied(
        &client,
        FETCH_OP_ID,
        json!({
            "namespace": "ns-1",
            "digest": digest_hex.clone(),
            "ranges": [
                { "offset": 0, "len": u64::MAX },
                { "offset": 0, "len": u64::MAX }
            ]
        }),
    )
    .await;
    match overflow_arm.result {
        Err(error) => {
            assert_eq!(error.code, "channel:open_failed", "{error:?}");
            assert!(
                error.message.contains("exceeds the total-bytes bound"),
                "the checked-add overflow arm is the same refusal: {error:?}"
            );
        }
        Ok(_) => panic!("an overflow-summed ranged open established: {overflow_arm:?}"),
    }

    // The bound is inclusive: lens summing exactly to the byte bound
    // establishes, and the served concatenation byte-verifies against
    // the store's own ranged reads (the ack digests ride as before).
    let (slice0, digest0) = store
        .read_range(&key.as_bytes(), 0, 8)
        .await
        .expect("the first ranged read");
    let (slice1, digest1) = store
        .read_range(&key.as_bytes(), 8, 8)
        .await
        .expect("the second ranged read");
    let (channel_id, reply, send, mut recv) = open_fetch(
        &client,
        "ns-1",
        &digest_hex,
        Some(json!([{ "offset": 0, "len": 8 }, { "offset": 8, "len": 8 }])),
    )
    .await
    .expect("the bound-exact ranged open establishes");
    assert_eq!(
        reply,
        json!({
            "channel_id": channel_id,
            "source": "store",
            "slice_digests": [hex::encode(digest0), hex::encode(digest1)]
        }),
        "the ack contract is unchanged (the slice digests ride the reply)"
    );
    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    let mut expected = slice0;
    expected.extend(slice1);
    assert_eq!(
        received, expected,
        "the bounded ranged request byte-verifies end to end"
    );
    drop(send);
}

/// The fanout seam's consumer side, live-broadcast arm (criterion:
/// a live broadcast tapped — the subscriber receives while the
/// producer runs): the entry never committed, so only the tap could
/// have served; the ack names the path (`source: "fanout"`).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_taps_a_live_broadcast_and_serves_while_the_producer_runs() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let seed: Vec<u8> = (0u8..16).collect();
    let digest = Digest::derive(&seed, seed.len() as u64);
    let key = Key::from(digest.clone());
    let digest_hex = hex::encode(digest.bytes());
    let publisher = ctx
        .fanout()
        .begin(&key, SlowSubscriberPolicy::Stall)
        .expect("the broadcast begins");

    let (channel_id, reply, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None)
        .await
        .expect("the fetch channel establishes");
    assert_eq!(
        reply,
        json!({ "channel_id": channel_id, "source": "fanout" }),
        "the tapped arm's ack: no length (a live put's tail), the path named"
    );
    assert!(
        !store.has(&key.as_bytes()).await.unwrap(),
        "no commit landed: only the tap could have served"
    );

    publisher.broadcast(Arc::new(seed[..8].to_vec())).await;
    let mut first = vec![0u8; 8];
    tokio::time::timeout(TEST_TIMEOUT, recv.read_exact(&mut first))
        .await
        .expect("the chunk reads")
        .expect("the chunk reads");
    publisher.broadcast(Arc::new(seed[8..].to_vec())).await;
    let mut second = vec![0u8; 8];
    tokio::time::timeout(TEST_TIMEOUT, recv.read_exact(&mut second))
        .await
        .expect("the chunk reads")
        .expect("the chunk reads");
    assert_eq!([first, second].concat(), seed, "the tapped feed, in order");

    publisher.finish();
    let mut rest = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut rest))
        .await
        .expect("the stream drains")
        .expect("the stream ends");
    assert!(rest.is_empty(), "the broadcast end closes the stream");
    drop(send);
}

/// The late joiner (criterion: a broadcast that ended degrades to
/// the post-commit verified `get` — asserted by which path served):
/// the broadcast begun and ended, the entry committed, the fetch
/// answers from the store path and byte-exact.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_late_joiner_after_the_broadcast_end_serves_the_store_path() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let seed: &[u8] = b"the late joiner's bytes";
    let (key, _pin) = store
        .put(Some(seed.len() as u64), std::io::Cursor::new(seed))
        .await
        .expect("seed put lands");
    let digest_key = key.clone();

    let publisher = ctx
        .fanout()
        .begin(&digest_key, SlowSubscriberPolicy::Drop)
        .expect("the broadcast begins");
    publisher.broadcast(Arc::new(b"stale feed".to_vec())).await;
    publisher.finish();

    let digest_hex = hex::encode(key.digest().bytes());
    let (channel_id, reply, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None)
        .await
        .expect("the fetch channel establishes");
    assert_eq!(
        reply,
        json!({ "channel_id": channel_id, "source": "store", "len": seed.len() }),
        "the ended broadcast's late joiner is the store path"
    );

    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    assert_eq!(received, seed, "identical bytes under CAS");
    drop(send);
}

/// The miss (criterion: the typed missing-error schema; the have/
/// need diff stays caller-side): an absent digest with no live
/// broadcast refuses establishment — `channel:open_failed` with
/// reason `unknown_resource`, the digest in the refusal's message,
/// the declared [`BLOB_MISSING`] contract on the spec.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_miss_refuses_establishment_with_the_typed_missing_shape() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let (spec, _establisher, _pump) = fetch_openables(Arc::clone(&ctx));
    assert!(
        spec.error_schemas
            .iter()
            .any(|error| error.code == BLOB_MISSING),
        "the declared missing-error contract is disclosed on the spec"
    );
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let absent = hex::encode(Digest::derive(b"never landed", 11).bytes());
    let refused = open_fetch(&client, "ns-1", &absent, None).await;
    match refused {
        Err(error) => {
            let call_error = error.call_error().expect("the wire error");
            assert_eq!(call_error.code, "channel:open_failed");
            assert_eq!(
                error.establishment_reason(),
                Some("unknown_resource"),
                "ADR-049's typed vocabulary for the absent resource"
            );
            assert!(
                call_error.message.contains(&absent),
                "the refusal carries the digest: {}",
                call_error.message
            );
        }
        Ok(_) => panic!("the absent digest refused, got a channel"),
    }
}

/// The read gate (criterion: closed-by-default, no read grant on
/// the namespace → denied): the ACL verdict is alkcall's, evaluated
/// at op entry ahead of the establishment phase.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_denies_the_ungranted_namespace_closed_by_default() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:stranger", identity_named("stranger"));
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("stranger")),
        )
        .await;

    let (digest_hex, _) = seeded_entry(&store, b"denied fetch").await;
    let denied = tokio::time::timeout(
        TEST_TIMEOUT,
        client.call_open_op(
            FETCH_OP_ID,
            json!({ "namespace": "ns-unlisted", "digest": digest_hex }),
        ),
    )
    .await
    .expect("the denied open resolves");
    assert_eq!(
        error_code(&denied),
        Some("FORBIDDEN"),
        "closed-by-default: a grant per namespace is the deferred cost"
    );
}

// ----------------------------------------------------------------
// blobs/put (the binary data plane's producer side): the verified
// Sink op through the channel halves — offer/stream/commit, the
// pin hand-over's custody split, and the fanout's producer side.
// ----------------------------------------------------------------

#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn open_put(
    client: &ChannelClient,
    input: Value,
) -> (
    u32,
    MpscSendStream,
    MpscRecvStream,
    alkcall::channels::client::PubOpenReply,
) {
    tokio::time::timeout(
        TEST_TIMEOUT,
        client.open_pub_channel(PUT_OP_ID, input, BLOBS_CHANNEL_ALPN),
    )
    .await
    .expect("the put's open resolves")
    .expect("the put channel opened")
}

/// A store-backed serving session: the context over `store`, the
/// harness carrying `blobs/have` beside it, alice's identity with
/// read+write on `ns-1`, and the channel client serving the put
/// openable. The put tests' common assembly.
#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn put_serving_with(
    store: &Arc<Store>,
    config: OpsConfig,
) -> (Arc<OpsContext>, OpsHarness, ChannelClient) {
    let ctx = Arc::new(OpsContext::new(Arc::clone(store), config));
    let harness = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx))]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let client = harness
        .spawn_channel_halves(
            vec![put_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;
    (ctx, harness, client)
}

#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn put_serving(store: &Arc<Store>) -> (Arc<OpsContext>, OpsHarness, ChannelClient) {
    put_serving_with(store, OpsConfig::default()).await
}

/// The known-length round trip (the op's core criterion): the
/// offer ack rides the merged response, the response's `digest`
/// matches the git derivation the putter carried (the byte-exact
/// standard — also the key the store committed under), the stored
/// content is byte-exact, and the minted token renews through
/// have's renewal form — the cross-task contract.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_known_length_round_trips_the_response_and_renews_via_have() {
    let store = test_store().await;
    let (ctx, harness, client) = put_serving(&store).await;

    let seed: &[u8] = b"the known-length put's bytes";
    let digest_hex = hex::encode(Digest::derive(seed, seed.len() as u64).bytes());
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    let (channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    assert!(channel_id > 0, "a data channel, not channel 0 itself");
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);

    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    let token_hex = out["token"].as_str().expect("the token field").to_owned();
    assert_eq!(
        out,
        json!({
            "channel_id": channel_id,
            "accepted": true,
            "token": token_hex,
            "digest": digest_hex,
        }),
        "both response fields always present (ADR-012 §2); the ack merges beside them"
    );

    assert!(
        store.has(&key.as_bytes()).await.expect("the has probe"),
        "the store committed under the git derivation"
    );
    let (stored_len, stored_blob) = store
        .get(&key.as_bytes())
        .await
        .expect("the read")
        .expect("present");
    assert_eq!(stored_len, seed.len() as u64);
    let mut stored = Vec::new();
    let mut pos = 0u64;
    while pos < stored_len {
        let (slice, _) = stored_blob.read_range(pos, 4096).await.expect("the window");
        stored.extend_from_slice(&slice);
        pos += slice.len() as u64;
    }
    assert_eq!(stored, seed, "byte-exact");

    // The hand-over's non-fleet custody (ADR-010 §4, ADR-012 §2):
    // the registry holds the put's guard — nothing expires.
    let token = PinToken::from_hex(&token_hex).expect("a hex token");
    assert!(ctx.tokens().holds(&token), "the registry took custody");
    assert_eq!(
        store.gc().pin_count(&key),
        1,
        "the registry's guard refcounts the entry"
    );

    // The cross-task contract: the minted token renews via have's
    // renewal form (the call halves serve the JSON plane).
    let call_client = harness.spawn_call_halves();
    let renewed = await_call(call_client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex], "token": token_hex }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        renewed.result.expect("the renewal probe"),
        json!({ "present": [true] }),
        "the put-minted token renews"
    );
}

/// The unknown-length round trip (the buffering path through the
/// wire): no `len` in the offer — the store's pre-threshold
/// buffering path carries it; the response's `digest` is what the
/// unknown-length putter needed (their later-op handle).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_unknown_length_round_trips_the_buffering_path() {
    let store = test_store().await;
    let (ctx, _harness, client) = put_serving(&store).await;

    let seed: Vec<u8> = (0u8..96).collect();
    let digest_hex = hex::encode(Digest::derive(&seed, seed.len() as u64).bytes());
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    let (_channel_id, mut send, recv, reply) =
        open_put(&client, json!({ "namespace": "ns-1" })).await;
    send.write_all(&seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);

    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    let token_hex = out["token"].as_str().expect("the token field").to_owned();
    assert_eq!(out["digest"], json!(digest_hex), "the canonical digest hex");
    assert!(
        out["accepted"].as_bool().expect("the ack"),
        "the offer ack rode in"
    );
    assert!(
        store.has(&key.as_bytes()).await.expect("the has probe"),
        "the put committed"
    );

    let token = PinToken::from_hex(&token_hex).expect("a hex token");
    assert!(ctx.tokens().holds(&token), "the registry took custody");
    assert_eq!(
        store.gc().pin_count(&key),
        1,
        "the registry holds the guard"
    );
}

/// The verification-failure shape (criterion: typed error, zero
/// residue on both tiers): a known-length offer whose stream dies
/// short — the large tier's staged form is exercised (the declared
/// length is over the threshold) — the store core rejects
/// `Verification`, the stage file is discarded, and no entry exists
/// on either tier.
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn put_verification_failure_leaves_zero_residue_on_both_tiers() {
    let unique = format!(
        "alkblobs-ops-put-residue-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let db = std::path::PathBuf::from(format!("/tmp/opencode/{unique}.db"));
    let root = std::path::PathBuf::from(format!("/tmp/opencode/{unique}"));
    let store = Arc::new(
        Store::dual_tier_sqlite_local(db, &root, 1024, std::time::Duration::ZERO)
            .await
            .expect("the dual-tier store constructs"),
    );
    let (_ctx, _harness, client) = put_serving(&store).await;

    let seed: Vec<u8> = (0u8..64).collect();
    let short: &[u8] = &seed[..32];

    let (_channel_id, mut send, recv, reply) =
        open_put(&client, json!({ "namespace": "ns-1", "len": 1024 })).await;
    send.write_all(short).await.expect("the prefix streams");
    drop(send);
    drop(recv);

    let failed_open = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect_err("a short stream against a known-length offer rejects");
    let failed = failed_open.call_error().expect("the wire error");
    assert_eq!(
        failed.code, VERIFICATION_REFUSED,
        "the canonical derivation's verification is the typed rejection: {failed:?}"
    );
    let putter_key = Key::from(
        Digest::from_git_oid_hex(&hex::encode(Digest::derive(short, 1024).bytes()))
            .expect("the derivation"),
    );
    assert!(
        !store
            .has(&putter_key.as_bytes())
            .await
            .expect("the has probe"),
        "the derived entry never committed"
    );
    let staging = root.join("staging");
    let stage_files = match std::fs::read_dir(&staging) {
        Ok(entries) => entries.count(),
        Err(_not_found) => 0,
    };
    assert_eq!(stage_files, 0, "zero residue on the large tier's staging");
}

/// The offered-digest mismatch (criterion: rejects before commit —
/// the known-length form's pre-commit check): the offer carries a
/// claim the honest stream's derivation contradicts; the response
/// is the typed mismatch (details name the claim and the derived
/// hex) and neither the claimed digest's key nor the honest
/// content's derived key ever committed.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_known_length_offer_digest_mismatch_rejects_before_commit() {
    let store = test_store().await;
    let (_ctx, _harness, client) = put_serving(&store).await;

    let seed: &[u8] = b"the honest stream's bytes";
    let honest_hex = hex::encode(Digest::derive(seed, seed.len() as u64).bytes());
    let claimed_hex = hex::encode(Digest::derive(b"unrelated claim", 16).bytes());

    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": claimed_hex, "len": seed.len() }),
    )
    .await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);

    let rejected_open = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect_err("the mismatched claim rejects");
    let rejected = rejected_open.call_error().expect("the wire error");
    assert_eq!(
        rejected.code, DIGEST_MISMATCH,
        "the offered digest is a claim; the derivation is the authority: {rejected:?}"
    );
    assert_eq!(
        rejected
            .details
            .as_ref()
            .and_then(|d| d.get("derived"))
            .and_then(|v| v.as_str()),
        Some(honest_hex.as_str()),
        "the details name the derivation that contradicted the claim"
    );
    for never_committed in [claimed_hex, honest_hex] {
        let key = Key::from(Digest::from_git_oid_hex(&never_committed).expect("the derivation"));
        assert!(
            !store.has(&key.as_bytes()).await.expect("the has probe"),
            "the rejection preceded the facade's commit: {never_committed}"
        );
    }
}

/// The offered-digest mismatch's unknown-length arm: the preamble is
/// only derivable at stream end (the prefix-less derivation pass is
/// the facade's own, over the restart-reached staged artifact), so
/// this claim rides the put's completion — the response still
/// rejects the put (nothing handed over), no token mints, and the
/// derived entry is valid CAS content the derivation itself
/// justifies (GC reaps an unreferenced one).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_unknown_length_offer_digest_mismatch_rejects_at_completion() {
    let store = test_store().await;
    let (_ctx, _harness, client) = put_serving(&store).await;

    let seed: &[u8] = b"the honest stream's bytes, unmeasured";
    let honest_hex = hex::encode(Digest::derive(seed, seed.len() as u64).bytes());
    let claimed_hex = hex::encode(Digest::derive(b"another unrelated", 18).bytes());

    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": claimed_hex }),
    )
    .await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);

    let rejected_open = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect_err("the mismatched claim rejects");
    let rejected = rejected_open.call_error().expect("the wire error");
    assert_eq!(rejected.code, DIGEST_MISMATCH, "{rejected:?}");
    assert_eq!(
        rejected
            .details
            .as_ref()
            .and_then(|d| d.get("derived"))
            .and_then(|v| v.as_str()),
        Some(honest_hex.as_str()),
        "the derivation is the authority, named in the details"
    );

    // The rejected put never handed over: no token rode the
    // response, no custody minted (the resolve's Err carries only
    // the mismatch).

    // The rejected put's guard dropped (the facade's put landed
    // unpinned to this handler); the honest content's derived entry
    // stands as valid CAS content.
    let honest_key = Key::from(Digest::from_git_oid_hex(&honest_hex).expect("the derivation"));
    assert!(
        store
            .has(&honest_key.as_bytes())
            .await
            .expect("the has probe"),
        "the facade's put committed (the derivation is the authority)"
    );
    assert_eq!(
        store.gc().pin_count(&honest_key),
        0,
        "no custody over the rejected put"
    );
}

/// The custody split's completion (the non-fleet side: the token is
/// the in-process guard's handle): the minted token's embedded
/// confirm converts through the ordered-step primitive — the held
/// guard's release and the converter's registration.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_token_custody_converts_at_the_confirmation() {
    let store = test_store().await;
    let (ctx, _harness, client) = put_serving(&store).await;
    let converted = Arc::new(super::tests::RecordingConverter {
        converted: Mutex::new(Vec::new()),
    });
    ctx.bind_converter("ns-1", converted);
    let seed: &[u8] = b"the custodial put's bytes";

    let (_channel_id, mut send, recv, reply) =
        open_put(&client, json!({ "namespace": "ns-1", "len": seed.len() })).await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    let token = PinToken::from_hex(out["token"].as_str().expect("the token")).expect("hex");
    let put_digest = out["digest"].as_str().expect("the digest").to_owned();
    let key = Key::from(Digest::from_git_oid_hex(&put_digest).expect("the derivation"));

    assert_eq!(
        store.gc().pin_count(&key),
        1,
        "the registry holds the guard"
    );
    assert!(
        ctx.confirm("ns-1", &token),
        "the embedded converter's registration consumes the held guard"
    );
    assert_eq!(
        store.gc().pin_count(&key),
        0,
        "the conversion released the held guard"
    );
    assert!(
        !ctx.tokens().holds(&token),
        "conversion-confirmed retires the mapping"
    );
}

/// The dedup re-put: a CAS dedup lands no new entry — the re-put's
/// response is `{token, digest}` again (the same digest), the
/// custodial custody re-mints (fresh guards refcounting the same
/// entry), and every re-put's token renews; the re-put's offered
/// digest (the putter re-offering the known one) passes the
/// cross-check by construction.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_dedup_re_put_re_mints_custody_and_both_tokens_renew() {
    let store = test_store().await;
    let (ctx, harness, client) = put_serving(&store).await;
    let seed: &[u8] = b"the re-put's identical bytes";
    let digest_hex = hex::encode(Digest::derive(seed, seed.len() as u64).bytes());
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    let mut tokens = Vec::new();
    for round in 0..3 {
        let offer = if round < 2 {
            json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() })
        } else {
            json!({ "namespace": "ns-1", "len": seed.len() })
        };
        let (_channel_id, mut send, recv, reply) = open_put(&client, offer).await;
        send.write_all(seed).await.expect("the bytes stream");
        drop(send);
        drop(recv);
        let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
            .await
            .expect("the re-put resolves")
            .expect("the re-put succeeds");
        assert_eq!(
            out["digest"],
            json!(digest_hex),
            "CAS dedup: the same digest answers (round {round})"
        );
        let token = PinToken::from_hex(out["token"].as_str().expect("the token")).expect("hex");
        tokens.push(token);
        assert!(
            store
                .stat(&key.as_bytes())
                .await
                .expect("the stat probe")
                .is_some(),
            "the entry exists (round {round})"
        );
    }
    assert_eq!(
        store.gc().pin_count(&key),
        3,
        "each re-put's custody re-mint refcounts the same entry"
    );
    for token in &tokens {
        assert!(
            ctx.tokens().holds(token),
            "each re-put minted its own custody"
        );
    }

    // The renewals answer present (the renewal form rides the
    // minted tokens; ADR-012 §2).
    let call_client = harness.spawn_call_halves();
    for token in &tokens {
        let renewed = await_call(call_client.call_with_payload(call_payload(
            "blobs/have",
            json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }),
            Some("token:alice"),
        )))
        .await;
        assert_eq!(
            renewed.result.expect("the renewal probe"),
            json!({ "present": [true] }),
            "the re-put renews the pin's custody form"
        );
    }
}

/// The admission refusal at the wire (ADR-019, the POC-L shape
/// reproduced against a mem store's serving path): with the registry
/// at its cap, the put's stream commits and verifies — and the
/// response is the declared `MINT_REFUSED` error over the
/// **committed** put, the digest in the details, no token issued, no
/// custody taken. Once headroom exists the CAS-idempotent re-put
/// answers `{token, digest}` again.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_mint_refused_at_the_registry_cap_answers_committed_with_no_token() {
    let store = test_store().await;
    let config = OpsConfig {
        token_registry_max_entries: 1,
        ..OpsConfig::default()
    };
    let (ctx, _harness, client) = put_serving_with(&store, config).await;
    let converted = Arc::new(RecordingConverter {
        converted: Mutex::new(Vec::new()),
    });
    ctx.bind_converter("ns-1", converted);

    // The loop's prior iteration: a minted, never-confirmed token
    // fills the one-entry registry.
    let loop_key = key_of(b"prior-loop-iteration");
    let pin = store.gc().acquire(&loop_key).await;
    let loop_token = ctx
        .mint_custodial("ns-1", loop_key, pin)
        .expect("the pre-cap mint");
    assert_eq!(ctx.tokens().len(), 1);

    let seed: &[u8] = b"the capped loop's conforming bytes";
    let digest_hex = hex::encode(Digest::derive(seed, seed.len() as u64).bytes());
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let refused = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect_err("the hand-over's mint refuses at the cap");
    let error = refused.call_error().expect("the wire error");
    assert_eq!(
        error.code, MINT_REFUSED,
        "the declared admission refusal: {error:?}"
    );
    assert_eq!(
        error
            .details
            .as_ref()
            .and_then(|d| d.get("digest"))
            .and_then(Value::as_str),
        Some(digest_hex.as_str()),
        "the committed digest rides the refusal (the putter's re-put anchor)"
    );

    // The honest answer's substance: the put committed; no custody
    // was taken; the refusal inserted nothing.
    assert!(
        store.has(&key.as_bytes()).await.expect("the has probe"),
        "the stream committed and verified on this node"
    );
    assert_eq!(
        store.gc().pin_count(&key),
        0,
        "no custody taken — the entry stands unreferenced, sweep eligible"
    );
    assert_eq!(
        ctx.tokens().len(),
        1,
        "the refusal inserted nothing (only the pre-cap mint is registered)"
    );

    // The recovery: headroom restored, the re-put (CAS-idempotent
    // over the committed content) answers {token, digest} again.
    assert!(
        ctx.confirm("ns-1", &loop_token),
        "the embedder's confirm retires the loop's entry and frees the headroom"
    );
    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let re_put = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the re-put resolves")
        .expect("the re-put succeeds");
    assert_eq!(re_put["digest"], json!(digest_hex), "CAS: the same digest");
    assert!(
        re_put["token"].as_str().is_some(),
        "a token answered this time"
    );
}

/// The ACL gate (criterion: closed-by-default, no write grant →
/// denied at op entry, ahead of the establishment phase).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn put_denies_the_ungranted_namespace_closed_by_default() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:stranger", identity_named("stranger"));
    let client = harness
        .spawn_channel_halves(
            vec![put_openable(Arc::clone(&ctx))],
            Some(identity_named("stranger")),
        )
        .await;

    let denied = tokio::time::timeout(
        TEST_TIMEOUT,
        client.call_open_op(PUT_OP_ID, json!({ "namespace": "ns-unlisted", "len": 4 })),
    )
    .await
    .expect("the denied open resolves");
    assert_eq!(
        error_code(&denied),
        Some("FORBIDDEN"),
        "closed-by-default: the write action needs its grant"
    );
}

/// The fanout seam's producer side, live (criterion:
/// serve-while-persisting — a fetch tapping the put's broadcast
/// serves while the put runs): the ack names the tapped path, the
/// feed is a contiguous nonempty suffix of the put's bytes (the tap
/// has no prefix — the A3 shape; the consumer's hash-check on a
/// partial feed is the degrade contract), and the put still
/// resolves `{token, digest}` with the commit landed.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fetch_taps_the_put_stream_and_serves_while_the_persisting_lands() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let client = harness
        .spawn_channel_halves(
            vec![
                put_openable(Arc::clone(&ctx)),
                fetch_openable(Arc::clone(&ctx)),
            ],
            Some(identity_named("alice")),
        )
        .await;

    let seed: Vec<u8> = (0u16..512).map(|i| (i % 251) as u8).collect::<Vec<u8>>();
    let digest_hex = hex::encode(Digest::derive(&seed, seed.len() as u64).bytes());

    // The putter streams its first bytes and holds: the put's
    // stream reader begins the broadcast (the offered digest keys
    // it); the commit cannot land (the stream is still open).
    let (channel_id, mut send, _recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(&seed[..256])
        .await
        .expect("the putter's first bytes stream");

    // The fetch taps the live broadcast. The put's open resolves
    // before its handler runs (the putter streams immediately), so
    // the first establish may race the broadcast's begin — a miss
    // answers unknown_resource (the entry is absent too) and the
    // bounded retry waits for the broadcast the put started; the
    // store path answering instead would mean the put already
    // committed (it cannot: the stream is still open).
    let (_fetch_id, fetch_reply, fetch_send, mut fetch_recv) = {
        let deadline = std::time::Instant::now() + TEST_TIMEOUT;
        loop {
            match open_fetch(&client, "ns-1", &digest_hex, None).await {
                Ok(fetch) => break fetch,
                Err(ref failure) if failure.establishment_reason() == Some("unknown_resource") => {
                    assert!(
                        std::time::Instant::now() < deadline,
                        "the put's broadcast never began"
                    );
                    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                    continue;
                }
                Err(failure) => {
                    panic!("the fetch establishment failed otherwise: {failure:?}")
                }
            }
        }
    };
    assert_eq!(
        fetch_reply["source"],
        json!("fanout"),
        "the tap: broadcast bytes serve while the put lands"
    );
    assert!(
        fetch_reply["len"].is_null(),
        "a live put's tail ack carries no length"
    );

    // The tap is the putter's coordination point: with the
    // subscriber arm live, the put's remaining bytes flow through
    // the tee's broadcast (Stall — the fetched consumer's reads
    // drive it) and the store arm concurrently.
    send.write_all(&seed[256..])
        .await
        .expect("the rest streams");
    drop(send);

    let drain = tokio::spawn(async move {
        let mut received = Vec::new();
        tokio::time::timeout(TEST_TIMEOUT, fetch_recv.read_to_end(&mut received))
            .await
            .expect("the fetched feed ends")
            .expect("the fetched feed ends");
        received
    });
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    assert_eq!(
        out["digest"],
        json!(digest_hex),
        "the commit's canonical digest"
    );
    let received = drain.await.expect("the fetched drain joins");
    assert!(
        !received.is_empty(),
        "the tap caught the put's remaining bytes at least"
    );
    assert_eq!(
        &seed[seed.len() - received.len()..],
        received.as_slice(),
        "the tap's feed is a contiguous in-order suffix (no prefix before the tap)"
    );
    assert!(
        store
            .has(
                &Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"))
                    .as_bytes()
            )
            .await
            .expect("the has probe"),
        "while the fetch served, the put landed"
    );
    drop(fetch_send);
    let _ = channel_id;
}

/// The kv-only mode's put-time threshold rejection (ADR-012 §6.2,
/// both forms): the known-length over-threshold offer rejects at
/// the put's start and the unknown-length over-flow rejects
/// mid-buffer — nothing commits either way, zero residue.
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn put_kv_only_threshold_rejects_both_offer_forms() {
    let db = std::path::PathBuf::from(format!(
        "/tmp/opencode/alkblobs-ops-put-kvonly-{}-{}.db",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    ));
    let store = Arc::new(
        Store::kv_only_sqlite(db, 128, std::time::Duration::ZERO)
            .await
            .expect("the kv-only store constructs"),
    );
    let (_ctx, _harness, client) = put_serving(&store).await;

    // The known-length over-threshold offer: rejected at the put's
    // start — the facade's put never entered the stream.
    let (_channel_id, mut send, recv, reply) =
        open_put(&client, json!({ "namespace": "ns-1", "len": 256 })).await;
    send.write_all(&[0u8; 256][..])
        .await
        .expect("the bytes stream");
    drop(send);
    drop(recv);
    let known_open = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect_err("the over-threshold known-length offer rejects");
    let known = known_open.call_error().expect("the wire error");
    assert_eq!(
        known.code, PUT_REFUSED,
        "the kv-only put-time rejection: {known:?}"
    );

    // The unknown-length form: rejected mid-buffer at the
    // threshold's overflow.
    let (_channel_id, mut send, recv, reply) =
        open_put(&client, json!({ "namespace": "ns-1" })).await;
    send.write_all(&[7u8; 256][..])
        .await
        .expect("the bytes stream");
    drop(send);
    drop(recv);
    let unknown_open = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect_err("the over-threshold unknown-length put rejects");
    let unknown = unknown_open.call_error().expect("the wire error");
    assert_eq!(unknown.code, PUT_REFUSED, "{unknown:?}");

    for content in [[0u8; 256], [7u8; 256]] {
        let key = Key::from(Digest::derive(&content, 256));
        assert!(
            !store.has(&key.as_bytes()).await.expect("the has probe"),
            "nothing committed"
        );
    }
}

// ----------------------------------------------------------------
// The JSON control plane's spec/handler pairs (ops-json-ops): the
// stat/have/delete families through the assembled registry.
// ----------------------------------------------------------------

#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn json_plane(store: Arc<Store>) -> (OpsHarness, Arc<OpsContext>) {
    let ctx = Arc::new(OpsContext::new(store, OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![
        stat_pair(Arc::clone(&ctx)),
        have_pair(Arc::clone(&ctx)),
    ]);
    (harness, ctx)
}

fn call_payload(op: &str, input: Value, auth_token: Option<&str>) -> Value {
    let mut payload = json!({
        "operationId": op,
        "input": input,
    });
    if let Some(token) = auth_token {
        payload["auth_token"] = json!(token);
    }
    payload
}

fn field_str<'a>(envelope: &'a ResponseEnvelope, code_field: &str) -> Option<&'a str> {
    match &envelope.result {
        Err(error) => error
            .details
            .as_ref()
            .and_then(|details| details.get(code_field))
            .and_then(Value::as_str),
        Ok(_) => None,
    }
}

fn error_code(envelope: &ResponseEnvelope) -> Option<&str> {
    match &envelope.result {
        Err(error) => Some(error.code.as_str()),
        Ok(_) => None,
    }
}

#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn seeded_entry(store: &Arc<Store>, seed: &[u8]) -> (String, Key) {
    let len = seed.len() as u64;
    let (key, _pin) = store
        .put(Some(len), std::io::Cursor::new(seed))
        .await
        .expect("seed put lands");
    let digest_hex = hex::encode(key.digest().bytes());
    (digest_hex, key)
}

/// `blobs/stat` (criterion: the stat round-trip + the miss shape):
/// a granted identity receives `{len}` for a seeded entry without a
/// content load (the read action's probe-before-offering shape); a
/// miss is the typed `BLOB_MISSING` error (not a `{len: null}`);
/// a malformed digest is rejected at the schema (`INVALID_INPUT`).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn stat_answers_len_the_miss_is_typed_and_the_digest_ingress_rejects() {
    let store = test_store().await;
    let (harness, _ctx) = json_plane(Arc::clone(&store)).await;
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness.spawn_call_halves();

    let (digest_hex, _) = seeded_entry(&store, b"stat round trip").await;

    let answer = await_call(client.call_with_payload(call_payload(
        "blobs/stat",
        json!({ "namespace": "ns-1", "digest": digest_hex }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        answer.result.unwrap(),
        json!({ "len": 15 }),
        "the seeded entry's length answers"
    );

    let missed = await_call(client.call_with_payload(call_payload(
        "blobs/stat",
        json!({ "namespace": "ns-1", "digest": "a".repeat(64) }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(error_code(&missed), Some(BLOB_MISSING));
    assert_eq!(
        field_str(&missed, "digest").map(str::to_owned),
        Some("a".repeat(64)),
        "the missing-error's details carry the digest"
    );

    let malformed = await_call(client.call_with_payload(call_payload(
        "blobs/stat",
        json!({ "namespace": "ns-1", "digest": "ZZZ-not-hex" }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        error_code(&malformed),
        Some("INVALID_INPUT"),
        "the schema gate rejects a malformed digest before the handler"
    );
}

/// `blobs/have` without the token (criterion: present/absent mixed
/// by index correspondence) and the read gate (closed-by-default):
/// an unlisted namespace's probe denies, the granted identity reads.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn have_answers_by_index_and_the_unlisted_namespace_denies() {
    let store = test_store().await;
    let (harness, _ctx) = json_plane(Arc::clone(&store)).await;
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness.spawn_call_halves();

    let (digest_hex, _) = seeded_entry(&store, b"have probe").await;
    let absent = hex::encode(Digest::derive(b"never put", 9).bytes());

    let mixed = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone(), absent.clone(), digest_hex.clone()] }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        mixed.result.unwrap(),
        json!({ "present": [true, false, true] }),
        "present[] corresponds to digests[] by index"
    );

    let denied = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-2-the-unlisted", "digests": [digest_hex.clone()] }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        error_code(&denied),
        Some("FORBIDDEN"),
        "closed-by-default: a grant per namespace is the deferred cost"
    );
}

/// `blobs/have`'s renewal form (criterion: the three answers, the
/// non-fleet arms): a live custodial row renews (`present: true`),
/// an unknown token answers `present: false` (never an error), a
/// mismatched digest answers `present: false`, the gate's case-
/// insensitive token pattern and the ingress's case-insensitive
/// decode agree (an uppercased wire form renews — S-5's case
/// posture), and a malformed token is the schema gate's
/// `INVALID_INPUT` at the wire (S-5's field bound) while the
/// handler's own ingress keeps the unknown-token mapping for calls
/// that ride past the gate. The arity
/// violation is the typed `INVALID_RENEWAL_FORM` — exercised at the
/// handler's ingress (the wire's schema gate would reject first).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_renewal_form_answers_and_the_ingress_rejects_the_arity_violation() {
    let store = test_store().await;
    let (harness, ctx) = json_plane(Arc::clone(&store)).await;
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness.spawn_call_halves();

    let (digest_hex, key) = seeded_entry(&store, b"renewal form").await;
    let other = Key::from(Digest::derive(b"another entry", 13));
    let pin = store.gc().acquire(&key).await;
    let token = ctx.mint_custodial("ns-1", key.clone(), pin).unwrap();
    let unknown_token = PinToken::generate().unwrap();

    let renewed = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        renewed.result.unwrap(),
        json!({ "present": [true] }),
        "a live custodial row renews from the held guard (nothing expires)"
    );

    let refused = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [hex::encode(other.digest().bytes())], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        refused.result.unwrap(),
        json!({ "present": [false] }),
        "a digest the token does not bind: the one-token-one-row check, present: false"
    );
    let foreign = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": unknown_token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        foreign.result.unwrap(),
        json!({ "present": [false] }),
        "a token this node did not mint: the validity-domain refusal, present: false, not an error"
    );

    let uppercased = token.to_hex().to_uppercase();
    let uppercased_wire = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": uppercased }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        uppercased_wire.result.unwrap(),
        json!({ "present": [true] }),
        "the gate's case-insensitive token pattern and the ingress's case-insensitive decode agree (S-5's case posture): an uppercased wire form renews"
    );

    let malformed_token = hex::encode(b"not-a-64-char-canonical-hex-token");
    let malformed_wire = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": malformed_token }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        error_code(&malformed_wire),
        Some("INVALID_INPUT"),
        "the schema gate's S-5 bound rejects the over-length token before any handler contact"
    );

    let (_, handler) = have_pair(Arc::clone(&ctx));
    let malformed = (handler)(
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": malformed_token }),
        bare_context(),
    )
    .await;
    assert_eq!(
        malformed.result.unwrap(),
        json!({ "present": [false] }),
        "a malformed token past the gate (direct/composed calls) is the unknown-token form, present: false, never an error"
    );

    let arity_wire = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone(), "0".repeat(64)], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        error_code(&arity_wire),
        Some("INVALID_INPUT"),
        "the schema gate enforces the renewal form's one-digest arity (if/then) at the wire"
    );

    let (_, handler) = have_pair(Arc::clone(&ctx));
    let arity = (handler)(
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone(), "0".repeat(64)], "token": token.to_hex() }),
        bare_context(),
    )
    .await;
    assert_eq!(
        error_code(&arity),
        Some(INVALID_RENEWAL_FORM),
        "the renewal form's one-digest arity is the handler's typed error"
    );

    let invalid = (handler)(
        json!({ "namespace": "ns-1", "digests": ["NOT-DIGEST-HEX"], "token": token.to_hex() }),
        bare_context(),
    )
    .await;
    assert_eq!(
        error_code(&invalid),
        Some(INVALID_DIGEST),
        "the digest ingress rejects, never guesses"
    );
}

/// The renewal's namespace parity (ADR-017 §3), server-less through
/// the wire: a renewal presented under a namespace other than the
/// mint record's answers the validity-domain refusal — `present:
/// false`, never an error, the mapping intact — and a subsequent
/// renew-under-the-minted-namespace succeeds (confirm's refusal
/// discipline is the parity's contrast; the `UnknownToken` mapping is
/// the decided outcome-mapping: a namespace-mismatched renewal is the
/// same "not this token's domain" answer confirm's refusal is).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn a_wrong_namespace_renewal_answers_the_validity_domain_refusal_and_the_mapping_stays_intact(
) {
    let store = test_store().await;
    let (harness, ctx) = json_plane(Arc::clone(&store)).await;
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    harness.grant("alice", "ns-2", &["read"]);
    let client = harness.spawn_call_halves();

    let (digest_hex, key) = seeded_entry(&store, b"namespace parity").await;
    let pin = store.gc().acquire(&key).await;
    let token = ctx.mint_custodial("ns-1", key.clone(), pin).unwrap();

    let mismatched = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-2", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        mismatched.result.unwrap(),
        json!({ "present": [false] }),
        "a renewal under a namespace the mint record contradicts: present: false, \
         not an error (ADR-017 §3's parity)"
    );

    let parity = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        parity.result.unwrap(),
        json!({ "present": [true] }),
        "the mapping stayed intact through the parity refusal: the renewal under the \
         minted namespace succeeds"
    );

    assert!(
        ctx.tokens().holds(&token),
        "the mapping survived the parity refusal and the renewal — confirm's refusal \
         discipline rides unmolested (the mapping retires only at a confirmed conversion)"
    );
}

/// `blobs/delete` (criterion: external discovery NOT_FOUND; the
/// internal operator-context composition deletes; the namespace
/// field is audit scoping, not gating — a grantless caller's
/// composed internal call runs under the operator authority):
/// deleted, fail-fast on the absent key, and `GcRefuse` as the
/// typed `GC_REFUSED`.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn delete_is_external_not_found_and_the_internal_authority_composition_deletes() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(store, OpsConfig::default()));
    let harness = OpsHarness::assemble_registrations(vec![
        OpsHarnessRegistration::pair(delete_pair(Arc::clone(&ctx)))
            .with_composition_authority(CompositionAuthority::new("blob-operator", [])),
        OpsHarnessRegistration::pair(composer_pair())
            .with_scoped_env(ScopedPeerEnv::new(["blobs/delete"])),
    ]);
    harness.with_token("token:stranger", identity_named("stranger"));
    let client = harness.spawn_call_halves();

    let external = await_call(client.call_with_payload(call_payload(
        "blobs/delete",
        json!({ "namespace": "ns-1", "digests": ["0".repeat(64)] }),
        Some("token:stranger"),
    )))
    .await;
    assert_eq!(
        error_code(&external),
        Some("NOT_FOUND"),
        "an Internal op is not discovered from the wire (the visibility discipline)"
    );

    let (digest_hex, key) = seeded_entry(ctx.store(), b"deleted by the operator").await;

    let composed = await_call(client.call_with_payload(call_payload(
        "blobs/compose-delete",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()] }),
        Some("token:stranger"),
    )))
    .await;
    assert_eq!(
        composed.result.unwrap(),
        json!({ "deleted": [digest_hex.clone()] }),
        "the operator-authority composition deletes; the namespace grant is not the gate"
    );
    assert!(
        !ctx.store().has(&key.as_bytes()).await.unwrap_or(true),
        "the entry is gone from the facade after the internal delete"
    );

    let (present_hex, present_key) = seeded_entry(ctx.store(), b"fail fast").await;
    let absent = hex::encode(Digest::derive(b"absent entry", 12).bytes());
    let fail_fast = await_call(client.call_with_payload(call_payload(
        "blobs/compose-delete",
        json!({ "namespace": "ns-1", "digests": [present_hex, absent.clone()] }),
        Some("token:stranger"),
    )))
    .await;
    assert_eq!(
        error_code(&fail_fast),
        Some(BLOB_MISSING),
        "the absent key is the typed missing error"
    );
    assert_eq!(
        field_str(&fail_fast, "digest").map(str::to_owned),
        Some(absent),
        "the missing-error's details carry the failing digest"
    );
    assert!(
        !ctx.store()
            .has(&present_key.as_bytes())
            .await
            .unwrap_or(true),
        "the fail-fast path's preceding digests deleted (nothing atomic was specced)"
    );

    let (pinned_hex, pinned_key) = seeded_entry(ctx.store(), b"protected entry").await;
    let _guard = ctx.store().gc().acquire(&pinned_key).await;
    let refused = await_call(client.call_with_payload(call_payload(
        "blobs/compose-delete",
        json!({ "namespace": "ns-1", "digests": [pinned_hex] }),
        Some("token:stranger"),
    )))
    .await;
    assert_eq!(
        error_code(&refused),
        Some(GC_REFUSED),
        "a protected key's refusal is the typed error"
    );
    assert_eq!(
        field_str(&refused, "digest").map(str::to_owned),
        Some(pinned_hex),
        "the refusal's details carry the protected digest"
    );
}

// ----------------------------------------------------------------
// The renewal form's fleet arms (the foundation's noted fleet
// forms) — against the live harness server (the fleet column's
// SKIP posture adapted in-crate): env-driven DSN, skipping cleanly
// server-less, acceptance green against the server.
// ----------------------------------------------------------------

/// The harness-server DSN (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB`
/// — the alkstore convention; host gated, the rest defaulting to
/// the harness server's values). `None` → the caller skips.
#[cfg(feature = "postgres")]
async fn fleet_dsn() -> Option<String> {
    let host = std::env::var("ALKBLOBS_PG_HOST")
        .ok()
        .filter(|v| !v.is_empty())?;
    let port = std::env::var("ALKBLOBS_PG_PORT")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "15432".to_owned());
    let user = std::env::var("ALKBLOBS_PG_USER")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "postgres".to_owned());
    let password = std::env::var("ALKBLOBS_PG_PASSWORD")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "poc".to_owned());
    let db = std::env::var("ALKBLOBS_PG_DB")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "blobs".to_owned());
    let dsn = format!("host={host} port={port} user={user} password={password} dbname={db}");
    match tokio::time::timeout(std::time::Duration::from_secs(3), async {
        let (client, connection) = tokio_postgres::connect(&dsn, tokio_postgres::NoTls)
            .await
            .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
        tokio::spawn(async move {
            let _ = connection.await;
        });
        Ok::<tokio_postgres::Client, Error>(client)
    })
    .await
    {
        Ok(Ok(_)) => Some(dsn),
        Ok(Err(e)) => {
            eprintln!("skipping (harness server unreachable): {e}");
            None
        }
        Err(_) => {
            eprintln!("skipping (harness server unreachable: connect timeout)");
            None
        }
    }
}

/// The pin-row instrument (the fleet column's medium-oracle
/// pattern, the have handler's renewal arm's observation): one
/// independent connection reads the schema's
/// `gc_pins(key, owner, expiry)` — the durable pin state — for one
/// key (`(owner, expiry)` per row; nanos are absolute wall time).
#[cfg(feature = "postgres")]
async fn fleet_pin_row(
    dsn: &str,
    schema: &str,
    key: &[u8],
) -> Result<Option<(Vec<u8>, i64)>, Error> {
    let (client, connection) = tokio_postgres::connect(dsn, tokio_postgres::NoTls)
        .await
        .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    let sql = format!(
        "SELECT owner, expiry FROM \"{}\".gc_pins WHERE key = $1",
        schema.replace('"', "\"\"")
    );
    let row = client
        .query_opt(&sql, &[&key])
        .await
        .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
    Ok(row.map(|row| {
        let owner: Vec<u8> = row.get(0);
        let expiry: i64 = row.get(1);
        (owner, expiry)
    }))
}

#[cfg(feature = "postgres")]
async fn fleet_schema_drop(dsn: &str, schema: &str) {
    if let Ok((admin, connection)) = tokio_postgres::connect(dsn, tokio_postgres::NoTls).await {
        tokio::spawn(async move {
            let _ = connection.await;
        });
        let sql = format!(
            "DROP SCHEMA IF EXISTS \"{}\" CASCADE",
            schema.replace('"', "\"\"")
        );
        let _ = admin.batch_execute(&sql).await;
    }
}

/// Two fleet-declared dual-tier stores over one freshly minted
/// schema (the fleet column's factory pattern, the have handler's
/// renewal arm's shape): the serving instance A mints the token,
/// the peer B plays the foreign mint — each with its own
/// store-instance id (the pin rows compose by owner) and the short
/// test pin TTL (the expiry horizon's observable cadence).
/// Returns the two stores; the caller owns the temp root and the
/// schema's teardown.
#[cfg(feature = "postgres")]
async fn open_fleet_pair(
    dsn: &str,
    schema: &str,
    root_a: &std::path::Path,
    root_b: &std::path::Path,
) -> Result<(Store, Store), Error> {
    let ttl = PAIR_TTL;
    let opts = |schema: &str| crate::PgOpts {
        schema: schema.to_owned(),
        ..crate::PgOpts::default()
    };
    let store_a = crate::Store::dual_tier_postgres_local_fleet(
        dsn,
        opts(schema),
        root_a.to_path_buf(),
        1024,
        std::time::Duration::ZERO,
        ttl,
        crate::FleetSharedMedia::asserted(),
    )
    .await?;
    let store_b = crate::Store::dual_tier_postgres_local_fleet(
        dsn,
        opts(schema),
        root_b.to_path_buf(),
        1024,
        std::time::Duration::ZERO,
        ttl,
        crate::FleetSharedMedia::asserted(),
    )
    .await?;
    Ok((store_a, store_b))
}

/// The have handler's renewal form against a **live fleet store**
/// (the foundation's noted fleet forms; ADR-012 §2's renew duty on
/// the wire): the namespace parity fires first (ADR-017 §3 — a
/// renewal under a namespace the mint record contradicts answers
/// `present: false`, not an error, and the mapping stays intact for
/// the correct-namespace renewal); the fleet token renews the live
/// row (`present: true` — the row's expiry moves, observed by the
/// independent connection's `gc_pins` read); a putter pausing past
/// the TTL answers `present: false` and **no renewal occurs** (the
/// expiry frozen at the last renewal — the never-resurrect
/// discipline, ADR-008 §2); a peer's foreign mint answers
/// `present: false`, not an error — the validity domain is the
/// minting node (ADR-012 §2). The custodial-mint-is-refused-under-
/// fleet arm rides the same contexts.
#[cfg(feature = "postgres")]
#[tokio::test]
async fn the_fleet_renewal_form_renews_the_live_row_and_never_resurrects_the_expired_one() {
    let Some(dsn) = fleet_dsn().await else {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
        return;
    };
    let schema = format!(
        "alkblobs_ops_have_renewal_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let root = std::env::temp_dir().join(format!(
        "alkblobs-ops-fleet-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    ));
    std::fs::create_dir_all(&root).expect("the temp root created");

    let (store_a, store_b) = open_fleet_pair(&dsn, &schema, &root.join("a"), &root.join("b"))
        .await
        .expect("the fleet pair constructs against the harness server");
    let store_a = Arc::new(store_a);
    let store_b = Arc::new(store_b);
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store_a), OpsConfig::default()));
    let peer = Arc::new(OpsContext::new(Arc::clone(&store_b), OpsConfig::default()));

    let harness = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx))]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    harness.grant("alice", "ns-2", &["read"]);
    let client = harness.spawn_call_halves();

    let seed: &[u8] = b"fleet renewal";
    let (key, pin_guard) = store_a
        .put(Some(seed.len() as u64), std::io::Cursor::new(seed))
        .await
        .expect("the fleet put commits the pin row");
    // The guard's drop at response completion is the put handler's
    // own act (ADR-010 §4) — from here the pin row's expiry is the
    // protection, and the TTL/3 renewal duty has stopped.
    drop(pin_guard);
    let digest_hex = hex::encode(key.digest().bytes());
    let owner = ctx
        .fleet_owner()
        .expect("the fleet-declared store's instance id");
    let token = ctx
        .mint_fleet("ns-1", key.clone(), &owner)
        .expect("the fleet mint over the serving instance");
    let minted_expiry = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the put's pin row");

    let custodial_refused: Pin = store_b.gc().acquire(&key).await;
    assert!(
        peer.mint_custodial("ns-1", key.clone(), custodial_refused)
            .is_err(),
        "the custodial mint is refused on a fleet-declared store (the row is the protection)"
    );

    // The namespace parity (ADR-017 §3) on the live fleet form: the
    // minted fleet token presented under a different namespace answers
    // present: false — the validity-domain refusal, not an error — and
    // the minted mapping stays intact for the correct-namespace renewal
    // that follows (the row's expiry moves only on that renewal).
    let parity = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-2", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        parity.result.expect("the parity probe answers"),
        json!({ "present": [false] }),
        "a renewal under a namespace the mint record contradicts: present: false, \
         not an error (ADR-017 §3)"
    );

    let renewed = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        renewed.result.expect("the renewal probe answers"),
        json!({ "present": [true] }),
        "the live row renews on the wire (the parity refusal retired nothing)"
    );
    let renewed_expiry = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the row");
    assert!(
        renewed_expiry.1 > minted_expiry.1,
        "the renewal moved the row's expiry forward (minted {}, renewed {})",
        minted_expiry.1,
        renewed_expiry.1
    );
    assert_eq!(
        renewed_expiry.0, owner,
        "the renewed row is the minting instance's"
    );

    let foreign = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": PinToken::generate().expect("the CSPRNG minted").to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        foreign.result.expect("the renewal probe answers"),
        json!({ "present": [false] }),
        "a token this node did not mint: the validity-domain refusal, present: false, not an error"
    );

    tokio::time::sleep(EXPIRY_HORIZON).await;
    let expired = await_call(client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        expired.result.expect("the renewal probe answers"),
        json!({ "present": [false] }),
        "a putter paused past the TTL: the row expired, present: false, no renewal"
    );
    let expired_expiry = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument");
    assert!(
        expired_expiry.is_none() || expired_expiry.map(|row| row.1) == Some(renewed_expiry.1),
        "the expired probe renewed nothing: the row's expiry froze at the last renewal, \
         and the row is gone only when a sweep's maintenance step reaped it"
    );

    fleet_schema_drop(&dsn, &schema).await;
    let _ = std::fs::remove_dir_all(&root);
    let _ = (store_a, store_b, ctx, peer);
}

/// The fleet put through the wire (the hand-over's fleet arm,
/// ADR-010 §4 + ops-surface.md's hand-over contract, on a live
/// fleet pair): the put commits the pin row, the response is
/// `{token, digest}`, the guard dropped at the response (pin_count
/// 0 — the row is the signal), the token renews via the renewal
/// form (the row's expiry moves), and the dedup re-put renews the
/// row again through the joint tx's upsert of `(key, owner)`.
#[cfg(feature = "postgres")]
#[tokio::test]
async fn put_fleet_round_trip_hands_the_row_over_and_the_re_put_renews_it() {
    let Some(dsn) = fleet_dsn().await else {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
        return;
    };
    let unique = format!(
        "{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let schema = format!("alkblobs_ops_put_fleet_{unique}");
    let root = std::env::temp_dir().join(format!("alkblobs-ops-fleet-put-{unique}"));
    std::fs::create_dir_all(&root).expect("the temp root created");

    let (store, _peer_store) = open_fleet_pair(&dsn, &schema, &root.join("a"), &root.join("b"))
        .await
        .expect("the fleet pair constructs against the harness server");
    let serving = Arc::new(store);
    let ctx = Arc::new(OpsContext::new(Arc::clone(&serving), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx))]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let client = harness
        .spawn_channel_halves(
            vec![put_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;
    let call_client = harness.spawn_call_halves();

    let seed: &[u8] = b"the fleet put's bytes";
    let digest_hex = hex::encode(Digest::derive(seed, seed.len() as u64).bytes());
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    // The first put: the row commits (the joint entry+pin tx), the
    // response carries both fields, and the guard drops at the
    // response — pin_count 0 is the row-is-the-signal observable.
    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    assert_eq!(out["digest"], json!(digest_hex), "the canonical digest hex");
    let token = PinToken::from_hex(out["token"].as_str().expect("the token field")).expect("hex");
    assert_eq!(
        serving.gc().pin_count(&key),
        0,
        "the guard dropped at the response: the row's expiry is the only protection"
    );
    let row_put = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the put's pin row");
    assert_eq!(
        row_put.0,
        ctx.fleet_owner().expect("the serving instance's id"),
        "the row's owner is the serving instance"
    );

    // The token renews via the wire's renewal form; the row's
    // expiry moves (the putter's TTL/3 cadence, ADR-012 §2).
    let renewed = await_call(call_client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        renewed.result.expect("the renewal probe answers"),
        json!({ "present": [true] }),
        "the put-minted fleet token renews"
    );
    let row_renewed = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the row");
    assert!(
        row_renewed.1 > row_put.1,
        "the renewal moved the row's expiry forward"
    );

    // The dedup re-put (CAS): the joint tx's upsert of (key, owner)
    // is the re-put's renewal form — the row's expiry moves again,
    // a fresh token mints, the putter's first token still maps.
    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the re-put resolves")
        .expect("the re-put succeeds");
    assert_eq!(
        out["digest"],
        json!(digest_hex),
        "CAS dedup: the same digest answers"
    );
    let re_token =
        PinToken::from_hex(out["token"].as_str().expect("the token field")).expect("hex");
    assert_ne!(re_token, token, "each hand-over mints its own token");
    assert!(ctx.tokens().holds(&re_token));
    assert!(
        ctx.tokens().holds(&token),
        "the first mapping outlives the re-put"
    );
    let row_after = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the row");
    assert!(
        row_after.1 > row_renewed.1,
        "the re-put's joint-tx upsert renewed the row"
    );

    fleet_schema_drop(&dsn, &schema).await;
    let _ = std::fs::remove_dir_all(&root);
    let _ = (serving, _peer_store, ctx, harness, client, call_client);
}

// ----------------------------------------------------------------
// The ops-e2e-flows capstone: the full lifecycle flows the
// handler tests exercise only in pieces, proven as a whole on
// the wire through this harness.
// ----------------------------------------------------------------

#[cfg(feature = "postgres")]
const EXPIRY_HORIZON: Duration = Duration::from_millis(3000);

/// The fleet pair's short test pin TTL — the widened load-tolerance
/// posture (the original 900 ms collapsed under a loaded-but-
/// unsaturated harness server: a starved cadence tick lands near the
/// expiry and keeps the row briefly alive past a tight horizon; every
/// waiting assertion now polls within its generous window instead,
/// and the TTL + horizon pair leaves the tick whole cadences of
/// margin).
#[cfg(feature = "postgres")]
const PAIR_TTL: Duration = Duration::from_millis(2000);

/// A set-backed liveness source (the rows fixtures' shape, the
/// flows edition): the store consults the same set the converter
/// registers into — the conversion hand-over's landing surface.
#[cfg(feature = "postgres")]
struct FlowLiveSet {
    live: Mutex<BTreeSet<Key>>,
}

#[cfg(feature = "postgres")]
impl Default for FlowLiveSet {
    fn default() -> FlowLiveSet {
        FlowLiveSet {
            live: Mutex::new(BTreeSet::new()),
        }
    }
}

#[cfg(feature = "postgres")]
impl FlowLiveSet {
    fn contains(&self, key: &Key) -> bool {
        self.live.lock().contains(key)
    }
}

#[cfg(feature = "postgres")]
impl crate::facade::LivenessSource for FlowLiveSet {
    fn live_keys(&self) -> Vec<Key> {
        self.live.lock().iter().cloned().collect()
    }
}

/// The namespace binding's converter in the hand-over flow's shape:
/// the registration act puts the digest into the embedder's
/// liveness registry (the registered source the protection hands
/// to).
#[cfg(feature = "postgres")]
struct HandoverConverter {
    live: Arc<FlowLiveSet>,
}

#[cfg(feature = "postgres")]
impl PinConverter for HandoverConverter {
    fn convert(&self, key: &Key) {
        self.live.live.lock().insert(key.clone());
    }
}

#[cfg(all(feature = "postgres", feature = "pg-lo"))]
async fn harness_table_count(dsn: &str, schema: &str, table: &str) -> usize {
    let (client, connection) = tokio_postgres::connect(dsn, tokio_postgres::NoTls)
        .await
        .expect("the instrument's connection");
    tokio::spawn(async move {
        let _ = connection.await;
    });
    let sql = format!(
        "SELECT count(*) FROM \"{}\".\"{}\"",
        schema.replace('"', "\"\""),
        table.replace('"', "\"\"")
    );
    let row = client.query_one(&sql, &[]).await.expect("the count");
    let count: i64 = row.get(0);
    count as usize
}

#[cfg(feature = "postgres")]
async fn renewal_answer(
    call_client: &alkcall::protocol::connection::CallConnection,
    namespace: &str,
    digest_hex: &str,
    token: &PinToken,
) -> Value {
    await_call(call_client.call_with_payload(call_payload(
        "blobs/have",
        json!({
            "namespace": namespace,
            "digests": [digest_hex],
            "token": token.to_hex()
        }),
        Some("token:alice"),
    )))
    .await
    .result
    .expect("the renewal probe answers")
}

#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn open_denied(client: &ChannelClient, op: &str, input: Value) -> ResponseEnvelope {
    tokio::time::timeout(TEST_TIMEOUT, client.call_open_op(op, input))
        .await
        .expect("the open resolves")
}

#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn have_call(
    call_client: &alkcall::protocol::connection::CallConnection,
    auth: &str,
    digests: &[String],
    token: Option<&PinToken>,
) -> Vec<bool> {
    let mut input = json!({ "namespace": "ns-1", "digests": digests });
    if let Some(token) = token {
        input["token"] = json!(token.to_hex());
    }
    let answer =
        await_call(call_client.call_with_payload(call_payload("blobs/have", input, Some(auth))))
            .await;
    match answer.result {
        Ok(out) => out["present"]
            .as_array()
            .expect("present[]")
            .iter()
            .map(|value| value.as_bool().expect("a bool"))
            .collect(),
        Err(error) => panic!("the have probe answers, got {error:?}"),
    }
}

/// The have/need flow, end to end: seed the store via the facade,
/// diff from the client side (the need announcement is the fetch
/// miss — the wire carries only concrete fetch requests), fetch
/// exactly the absent set after the putter fills it, re-diff to
/// all-present. The renewal-cadence boundary rides along (the test
/// plays the embedder's cadence explicitly, ADR-013 §1).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_have_need_flow_diffs_caller_side_and_fetches_exactly_the_need_set() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![
        stat_pair(Arc::clone(&ctx)),
        have_pair(Arc::clone(&ctx)),
    ]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let fetch_client = harness
        .spawn_channel_halves(
            vec![
                fetch_openable(Arc::clone(&ctx)),
                put_openable(Arc::clone(&ctx)),
            ],
            Some(identity_named("alice")),
        )
        .await;
    let call_client = harness.spawn_call_halves();

    let (d1_hex, _) = seeded_entry(&store, b"have/need seed one").await;
    let (d2_hex, _) = seeded_entry(&store, b"have/need seed two").await;
    let d3_seed: Vec<u8> = (0u8..32).collect();
    let d3_hex = hex::encode(Digest::derive(&d3_seed, d3_seed.len() as u64).bytes());
    let manifest = vec![d1_hex.clone(), d2_hex.clone(), d3_hex.clone()];

    let round_one = have_call(&call_client, "token:alice", &manifest, None).await;
    assert_eq!(
        round_one,
        vec![true, true, false],
        "the diff is caller-side: the wire answered the digests' presence, \
         the need set fell out of it"
    );
    let need_hex = manifest[2].clone();

    let announcement = open_fetch(&fetch_client, "ns-1", &need_hex, None).await;
    match announcement {
        Err(error) => {
            assert_eq!(
                error.establishment_reason(),
                Some("unknown_resource"),
                "the need announcement is the fetch miss: the wire carried a concrete \
                 fetch request and the absent digest refused"
            );
        }
        Ok(_) => panic!("the absent digest refused, got a channel"),
    }
    let no_need_op = await_call(call_client.call_with_payload(call_payload(
        "blobs/need",
        json!({ "namespace": "ns-1", "digests": manifest }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        error_code(&no_need_op),
        Some("NOT_FOUND"),
        "no need op exists: the diff is caller-side, the wire carries only concrete \
         fetch requests"
    );

    let (_channel_id, mut send, recv, reply) = open_put(
        &fetch_client,
        json!({ "namespace": "ns-1", "digest": need_hex.clone(), "len": d3_seed.len() }),
    )
    .await;
    send.write_all(&d3_seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    assert_eq!(out["digest"], json!(need_hex), "the put filled the need");
    let putter_token = PinToken::from_hex(out["token"].as_str().expect("the token")).expect("hex");
    let renewal = have_call(
        &call_client,
        "token:alice",
        std::slice::from_ref(&need_hex),
        Some(&putter_token),
    )
    .await;
    assert_eq!(renewal, vec![true], "the putter's cadence rode the renewal form (the test drives renewal explicitly — no in-crate renewal worker exists, ADR-013 §1)");

    let round_two = have_call(&call_client, "token:alice", &manifest, None).await;
    assert_eq!(
        round_two,
        vec![true, true, true],
        "re-diffed after the fill"
    );

    let (channel_id, ack, send, mut recv) = open_fetch(&fetch_client, "ns-1", &need_hex, None)
        .await
        .expect("the need-set fetch establishes");
    assert_eq!(
        ack,
        json!({"channel_id": channel_id, "source": "store", "len": d3_seed.len()}),
        "the wire carried a concrete fetch request for exactly the need set"
    );
    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    assert_eq!(received, d3_seed, "byte-exact");
    assert_eq!(
        Key::from(Digest::derive(&received, received.len() as u64)),
        Key::from(Digest::from_git_oid_hex(&need_hex).expect("the derivation")),
        "the consumer's whole-blob hash-check passed"
    );
    drop(send);
}

/// The fleet token lifecycle's success half, end to end against the
/// harness server (env-driven, skipping cleanly server-less): the
/// wire put mints over the committed row, the renewal form renews
/// before expiry (the row's expiry observed moving), the
/// embedder's registration confirms and the row's protection hands
/// to the registered liveness source (the namespace binding's
/// converter), and the row expires naturally — the entry survives
/// on the registered source's protection.
#[cfg(feature = "postgres")]
#[tokio::test]
async fn the_fleet_token_lifecycle_hands_over_to_the_registered_liveness_source() {
    let Some(dsn) = fleet_dsn().await else {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
        return;
    };
    let unique = format!(
        "{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let schema = format!("alkblobs_flows_fleet_success_{unique}");
    let root = std::env::temp_dir().join(format!("alkblobs-flows-fleet-success-{unique}"));
    std::fs::create_dir_all(&root).expect("the temp root created");

    let (store, _peer) = open_fleet_pair(&dsn, &schema, &root.join("a"), &root.join("b"))
        .await
        .expect("the fleet pair constructs against the harness server");
    let serving = Arc::new(store);
    let ctx = Arc::new(OpsContext::new(Arc::clone(&serving), OpsConfig::default()));

    let live = Arc::new(FlowLiveSet::default());
    let _registration = serving
        .register_liveness_source(crate::facade::LivenessRegistration::InProcess(
            Arc::clone(&live) as Arc<dyn crate::facade::LivenessSource>,
        ))
        .expect("the embedder's liveness registration");
    let converter = Arc::new(HandoverConverter {
        live: Arc::clone(&live),
    });
    ctx.bind_converter("ns-1", converter as Arc<dyn PinConverter>);

    let harness = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx))]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let client = harness
        .spawn_channel_halves(
            vec![put_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;
    let call_client = harness.spawn_call_halves();

    let seed: &[u8] = b"the fleet lifecycle put";
    let digest_hex = hex::encode(Digest::derive(seed, seed.len() as u64).bytes());
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    assert_eq!(out["digest"], json!(digest_hex));
    let token = PinToken::from_hex(out["token"].as_str().expect("the token field")).expect("hex");

    let owner = ctx.fleet_owner().expect("the serving instance's id");
    let row_minted = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the put's pin row");
    assert_eq!(
        row_minted.0, owner,
        "the row's owner is the serving instance"
    );
    assert_eq!(
        serving.gc().pin_count(&key),
        0,
        "the guard dropped at the response: the row's expiry is the protection"
    );

    assert_eq!(
        renewal_answer(&call_client, "ns-1", &digest_hex, &token).await,
        json!({ "present": [true] }),
        "renewed before expiry"
    );
    let row_renewed_1 = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the row");
    assert!(
        row_renewed_1.1 > row_minted.1,
        "the renewal moved the row's expiry forward"
    );
    assert_eq!(
        renewal_answer(&call_client, "ns-1", &digest_hex, &token).await,
        json!({ "present": [true] })
    );
    let row_renewed_2 = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the row");
    assert!(
        row_renewed_2.1 > row_renewed_1.1,
        "the cadence moved the expiry again (the test plays the embedder's TTL/3 role)"
    );

    assert!(
        ctx.confirm("ns-1", &token),
        "the embedder's registration confirms: the namespace binding's converter ran"
    );
    assert!(
        !ctx.tokens().holds(&token),
        "conversion-confirmed retires the mapping"
    );
    assert!(
        live.contains(&key),
        "the converter registered the key into the embedder's liveness registry"
    );
    assert_eq!(
        renewal_answer(&call_client, "ns-1", &digest_hex, &token).await,
        json!({ "present": [false] }),
        "the row's protection handed to the registered source: the token is spent, \
         the probe answers present: false, never an error"
    );
    assert!(
        serving.has(&key.as_bytes()).await.expect("the has probe"),
        "the entry stands after the registration confirms"
    );

    tokio::time::sleep(EXPIRY_HORIZON).await;
    let staged = serving.sweep().await.expect("sweeps are never errors");
    assert_eq!(staged.aborted, None);
    let row_after_staging = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument");
    assert!(
        row_after_staging.is_none(),
        "the expired row reaped by the sweep's maintenance step"
    );
    let executed = serving.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (executed.deleted, executed.aborted),
        (0, None),
        "the registered source's protection held: nothing deleted"
    );
    assert!(
        serving.has(&key.as_bytes()).await.expect("the has probe"),
        "the entry survives on the registered liveness source"
    );
    assert!(live.contains(&key), "the source carried the key throughout");

    fleet_schema_drop(&dsn, &schema).await;
    let _ = std::fs::remove_dir_all(&root);
    let _ = (
        serving,
        ctx,
        harness,
        client,
        call_client,
        _peer,
        _registration,
    );
}

/// The fleet token lifecycle's failure half against the harness
/// server: pause past TTL → the renewal probe answers
/// `present: false` → the sweeper reaps the row and the sweep
/// deletes the unprotected entry → the re-put backstop (the
/// digest re-offered; CAS makes the re-put idempotent) recovers
/// byte-identically. The race resolves to a re-put, never to
/// silent loss (ADR-010 §4's expiry semantics verbatim, on the
/// wire).
#[cfg(feature = "postgres")]
#[tokio::test]
async fn the_paused_fleet_token_expires_reaps_and_the_reput_backstop_never_loses_silently() {
    let Some(dsn) = fleet_dsn().await else {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
        return;
    };
    let unique = format!(
        "{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let schema = format!("alkblobs_flows_fleet_failure_{unique}");
    let root = std::env::temp_dir().join(format!("alkblobs-flows-fleet-failure-{unique}"));
    std::fs::create_dir_all(&root).expect("the temp root created");

    let (store, _peer) = open_fleet_pair(&dsn, &schema, &root.join("a"), &root.join("b"))
        .await
        .expect("the fleet pair constructs against the harness server");
    let serving = Arc::new(store);
    let ctx = Arc::new(OpsContext::new(Arc::clone(&serving), OpsConfig::default()));
    let _registration = serving
        .register_liveness_source(crate::facade::LivenessRegistration::InProcess(Arc::new(
            FlowLiveSet::default(),
        )
            as Arc<dyn crate::facade::LivenessSource>))
        .expect("the embedder's liveness registration");

    let harness = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx))]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let client = harness
        .spawn_channel_halves(
            vec![
                put_openable(Arc::clone(&ctx)),
                fetch_openable(Arc::clone(&ctx)),
            ],
            Some(identity_named("alice")),
        )
        .await;
    let call_client = harness.spawn_call_halves();

    let seed: Vec<u8> = (0u8..48).map(|i| i % 31).collect();
    let digest_hex = hex::encode(Digest::derive(&seed, seed.len() as u64).bytes());
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(&seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    let token = PinToken::from_hex(out["token"].as_str().expect("the token field")).expect("hex");
    let row_put = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the put's pin row");

    // The putter's pause past TTL — the load-tolerance posture: the
    // wait polls the **row instrument** until the expiry is past
    // instead of sampling once at a fixed wall time (a starved
    // harness server stretches everything; a poll on the have probe
    // itself would feed the row it watches — every present: true
    // answer is a renewal). Then the single have call asserts the
    // expired answer.
    let pause_deadline = std::time::Instant::now() + Duration::from_secs(10);
    loop {
        let row = fleet_pin_row(&dsn, &schema, &key.as_bytes())
            .await
            .expect("the row instrument")
            .expect("the row stands until a sweep's maintenance reaps it");
        if row.1 <= crate::pg_common::time_to_nanos(std::time::SystemTime::now()) {
            break;
        }
        assert!(
            std::time::Instant::now() < pause_deadline,
            "the putter's pause past TTL: the row's expiry is past"
        );
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    let expired = await_call(call_client.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        expired.result.expect("the renewal probe answers"),
        json!({ "present": [false] }),
        "the putter paused past TTL: present: false, no renewal occurred"
    );
    let row_after_probe = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument");
    assert_eq!(
        row_after_probe.map(|row| row.1),
        Some(row_put.1),
        "the expired probe renewed nothing: the row's expiry froze at the last write"
    );

    let staged = serving.sweep().await.expect("sweeps are never errors");
    assert_eq!(staged.aborted, None);
    assert_eq!(
        staged.candidates_staged, 1,
        "the unprotected entry staged for the delete window"
    );
    assert!(
        fleet_pin_row(&dsn, &schema, &key.as_bytes())
            .await
            .expect("the row instrument")
            .is_none(),
        "the sweeper reaped the expired row (the maintenance step)"
    );
    let executed = serving.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (executed.deleted, executed.aborted),
        (1, None),
        "the execution deleted the unprotected entry exactly once"
    );
    assert!(
        !serving.has(&key.as_bytes()).await.expect("the has probe"),
        "the entry did not survive the sweep"
    );

    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": seed.len() }),
    )
    .await;
    send.write_all(&seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the re-put resolves")
        .expect("the re-put succeeds");
    assert_eq!(
        out["digest"],
        json!(digest_hex),
        "CAS dedup: the re-offered digest recovered the same canonical address"
    );
    let re_token =
        PinToken::from_hex(out["token"].as_str().expect("the token field")).expect("hex");
    let row_recovered = fleet_pin_row(&dsn, &schema, &key.as_bytes())
        .await
        .expect("the row instrument")
        .expect("the re-put's pin row");
    assert_eq!(
        row_recovered.0,
        ctx.fleet_owner().expect("the serving instance's id"),
        "the recovery restored the pool's own protection"
    );
    let mut renewal;
    let renew_deadline = std::time::Instant::now() + Duration::from_millis(1500);
    loop {
        renewal = await_call(call_client.call_with_payload(call_payload(
            "blobs/have",
            json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": re_token.to_hex() }),
            Some("token:alice"),
        )))
        .await
        .result
        .expect("the renewal probe answers");
        if renewal == json!({ "present": [true] }) {
            break;
        }
        assert!(
            std::time::Instant::now() < renew_deadline,
            "the recovery's token renews (the closed-degradation fold on a transient \
             harness error re-asks within the TTL cadence before it counts)"
        );
        tokio::time::sleep(Duration::from_millis(250)).await;
    }
    assert_eq!(
        renewal,
        json!({ "present": [true] }),
        "the recovery's token renews (the closed-degradation fold on a transient \
         harness error re-asks within the TTL cadence before it counts)"
    );

    let (channel_id, ack, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None)
        .await
        .expect("the recovered entry fetches");
    assert_eq!(
        ack,
        json!({ "channel_id": channel_id, "source": "store", "len": seed.len() }),
        "the recovered entry serves: never a silent loss"
    );
    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    assert_eq!(received, seed, "the recovery is byte-identical");
    drop(send);

    fleet_schema_drop(&dsn, &schema).await;
    let _ = std::fs::remove_dir_all(&root);
    let _ = (
        serving,
        ctx,
        harness,
        client,
        call_client,
        _peer,
        _registration,
    );
}

/// The foreign-token refusal: a token minted by node A presented to
/// node B (a second fleet instance's ops module over the same pool)
/// answers `present: false`, not an error — the validity domain is
/// the minting node (ADR-012 §2); B's own mint over the same pool
/// renews.
#[cfg(feature = "postgres")]
#[tokio::test]
async fn the_foreign_token_refusal_is_present_false_not_an_error() {
    let Some(dsn) = fleet_dsn().await else {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
        return;
    };
    let unique = format!(
        "{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let schema = format!("alkblobs_flows_foreign_token_{unique}");
    let root = std::env::temp_dir().join(format!("alkblobs-flows-foreign-token-{unique}"));
    std::fs::create_dir_all(&root).expect("the temp root created");

    let (store_a, store_b) = open_fleet_pair(&dsn, &schema, &root.join("a"), &root.join("b"))
        .await
        .expect("the fleet pair constructs against the harness server");
    let node_a = Arc::new(store_a);
    let node_b = Arc::new(store_b);
    let ctx_a = Arc::new(OpsContext::new(Arc::clone(&node_a), OpsConfig::default()));
    let ctx_b = Arc::new(OpsContext::new(Arc::clone(&node_b), OpsConfig::default()));

    let harness_a = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx_a))]);
    harness_a.with_token("token:peer", identity_named("peer"));
    harness_a.grant("peer", "ns-1", &["read"]);
    let call_a = harness_a.spawn_call_halves();
    let harness_b = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx_b))]);
    harness_b.with_token("token:peer", identity_named("peer"));
    harness_b.grant("peer", "ns-1", &["read"]);
    let call_b = harness_b.spawn_call_halves();

    let seed: &[u8] = b"the foreign token's entry";
    let (key, _pin) = node_a
        .put(Some(seed.len() as u64), std::io::Cursor::new(seed))
        .await
        .expect("the put commits the pin row");
    let digest_hex = hex::encode(key.digest().bytes());
    let owner_a = ctx_a.fleet_owner().expect("node A's instance id");
    let token_a = ctx_a
        .mint_fleet("ns-1", key.clone(), &owner_a)
        .expect("node A mints over its own row");
    drop(_pin);

    let (_, pin_b) = node_b
        .put(Some(seed.len() as u64), std::io::Cursor::new(seed))
        .await
        .expect("node B's re-put composes its own pin row");
    let owner_b = ctx_b.fleet_owner().expect("node B's instance id");
    let token_b = ctx_b
        .mint_fleet("ns-1", key.clone(), &owner_b)
        .expect("node B mints over its own row");
    drop(pin_b);

    let foreign = await_call(call_b.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token_a.to_hex() }),
        Some("token:peer"),
    )))
    .await;
    assert_eq!(
        foreign.result.expect("the renewal probe answers"),
        json!({ "present": [false] }),
        "a token minted by node A presented to node B: present: false, not an error \
         (the validity domain is the minting node, ADR-012 §2)"
    );
    let reverse = await_call(call_a.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token_b.to_hex() }),
        Some("token:peer"),
    )))
    .await;
    assert_eq!(
        reverse.result.expect("the renewal probe answers"),
        json!({ "present": [false] }),
        "the refusal is symmetric: B's token is foreign to A"
    );
    let own = await_call(call_b.call_with_payload(call_payload(
        "blobs/have",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token_b.to_hex() }),
        Some("token:peer"),
    )))
    .await;
    assert_eq!(
        own.result.expect("the renewal probe answers"),
        json!({ "present": [true] }),
        "node B's own mint over the same pool renews"
    );

    fleet_schema_drop(&dsn, &schema).await;
    let _ = std::fs::remove_dir_all(&root);
    let _ = (node_a, node_b, ctx_a, ctx_b, harness_a, harness_b, call_a);
}

/// The non-fleet token lifecycle, server-less: the wire put
/// custodially mints (the registry holds the guard), have-present
/// answers from the custody, the embedder's registration confirms
/// and the guard converts (the namespace binding's converter), and
/// the token retires — with the entry standing after conversion.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_non_fleet_lifecycle_custody_confirmation_conversion_and_retirement() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx))]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let client = harness
        .spawn_channel_halves(
            vec![put_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;
    let call_client = harness.spawn_call_halves();

    let seed: &[u8] = b"the non-fleet lifecycle put";
    let (_channel_id, mut send, recv, reply) =
        open_put(&client, json!({ "namespace": "ns-1", "len": seed.len() })).await;
    send.write_all(seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the put succeeds");
    let digest_hex = out["digest"].as_str().expect("the digest field").to_owned();
    let token = PinToken::from_hex(out["token"].as_str().expect("the token field")).expect("hex");
    let key = Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"));

    assert_eq!(
        store.gc().pin_count(&key),
        1,
        "the registry holds the guard (the custodial form)"
    );
    assert_eq!(
        have_call(
            &call_client,
            "token:alice",
            std::slice::from_ref(&digest_hex),
            Some(&token)
        )
        .await,
        vec![true],
        "have-present answers from the custody (nothing expires)"
    );

    let converted = Arc::new(super::tests::RecordingConverter {
        converted: Mutex::new(Vec::new()),
    });
    ctx.bind_converter("ns-1", converted.clone());
    assert!(
        ctx.confirm("ns-1", &token),
        "the embedder's registration confirms: the guard converts through the \
         ordered-step primitive"
    );
    assert_eq!(
        store.gc().pin_count(&key),
        0,
        "the conversion consumed the held guard"
    );
    assert!(
        !ctx.tokens().holds(&token),
        "conversion-confirmed retires the mapping"
    );
    assert_eq!(
        converted.captured(),
        vec![key.clone()],
        "the converter received the digest"
    );
    assert_eq!(
        have_call(
            &call_client,
            "token:alice",
            std::slice::from_ref(&digest_hex),
            Some(&token)
        )
        .await,
        vec![false],
        "the retired token's probe answers present: false, never an error"
    );
    assert_eq!(
        have_call(
            &call_client,
            "token:alice",
            std::slice::from_ref(&digest_hex),
            None
        )
        .await,
        vec![true],
        "the pure probe answers: the entry stands on the registered source's \
         protection (the hand-over contract's have-present observable)"
    );
}

/// The ACL matrix, end to end: closed-by-default (the unlisted
/// namespace denies read AND write); the read/write split (the
/// read-granted identity fetches/stats/haves but cannot put; the
/// write-granted identity puts and neither it nor a peer granted
/// elsewhere can fetch); delete only under the internal authority
/// context.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_acl_matrix_closed_by_default_the_read_write_split_and_internal_only_delete() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble_registrations(vec![
        OpsHarnessRegistration::pair(stat_pair(Arc::clone(&ctx))),
        OpsHarnessRegistration::pair(have_pair(Arc::clone(&ctx))),
        OpsHarnessRegistration::pair(delete_pair(Arc::clone(&ctx)))
            .with_composition_authority(CompositionAuthority::new("blob-operator", [])),
        OpsHarnessRegistration::pair(composer_pair())
            .with_scoped_env(ScopedPeerEnv::new(["blobs/delete"])),
    ]);
    for (token, id) in [
        ("token:reader", "reader"),
        ("token:writer", "writer"),
        ("token:other", "other"),
        ("token:stranger", "stranger"),
    ] {
        harness.with_token(token, identity_named(id));
    }
    harness.grant("reader", "ns-1", &["read"]);
    harness.grant("writer", "ns-1", &["write"]);
    harness.grant("other", "ns-2", &["read"]);
    let call_client = harness.spawn_call_halves();
    let reader_client = harness
        .spawn_channel_halves(
            vec![
                put_openable(Arc::clone(&ctx)),
                fetch_openable(Arc::clone(&ctx)),
            ],
            Some(identity_named("reader")),
        )
        .await;
    let writer_client = harness
        .spawn_channel_halves(
            vec![
                put_openable(Arc::clone(&ctx)),
                fetch_openable(Arc::clone(&ctx)),
            ],
            Some(identity_named("writer")),
        )
        .await;
    let stranger_client = harness
        .spawn_channel_halves(
            vec![
                put_openable(Arc::clone(&ctx)),
                fetch_openable(Arc::clone(&ctx)),
            ],
            Some(identity_named("stranger")),
        )
        .await;
    let other_client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("other")),
        )
        .await;

    let seed: &[u8] = b"the ACL matrix's entry";
    let (key, _seed_pin) = store
        .put(Some(seed.len() as u64), std::io::Cursor::new(seed))
        .await
        .expect("seed put lands");
    let digest_hex = hex::encode(key.digest().bytes());

    assert_eq!(
        error_code(
            &await_call(call_client.call_with_payload(call_payload(
                "blobs/stat",
                json!({ "namespace": "ns-unlisted", "digest": digest_hex.clone() }),
                Some("token:stranger"),
            )))
            .await
        ),
        Some("FORBIDDEN"),
        "closed-by-default: the unlisted namespace denies read (stat)"
    );
    assert_eq!(
        error_code(
            &await_call(call_client.call_with_payload(call_payload(
                "blobs/have",
                json!({ "namespace": "ns-unlisted", "digests": [digest_hex.clone()] }),
                Some("token:stranger"),
            )))
            .await
        ),
        Some("FORBIDDEN"),
        "closed-by-default: the unlisted namespace denies read (have)"
    );
    assert_eq!(
        error_code(
            &open_denied(
                &stranger_client,
                FETCH_OP_ID,
                json!({ "namespace": "ns-unlisted", "digest": digest_hex.clone() }),
            )
            .await
        ),
        Some("FORBIDDEN"),
        "closed-by-default: the unlisted namespace denies read (fetch)"
    );
    assert_eq!(
        error_code(
            &open_denied(
                &stranger_client,
                PUT_OP_ID,
                json!({ "namespace": "ns-unlisted", "len": 4 }),
            )
            .await
        ),
        Some("FORBIDDEN"),
        "closed-by-default: the unlisted namespace denies write"
    );

    let stat = await_call(call_client.call_with_payload(call_payload(
        "blobs/stat",
        json!({ "namespace": "ns-1", "digest": digest_hex.clone() }),
        Some("token:reader"),
    )))
    .await;
    assert_eq!(
        stat.result.unwrap(),
        json!({ "len": seed.len() }),
        "the read-granted identity stats"
    );
    assert_eq!(
        have_call(
            &call_client,
            "token:reader",
            std::slice::from_ref(&digest_hex),
            None
        )
        .await,
        vec![true],
        "the read-granted identity haves"
    );
    let (_channel_id, ack, send, mut recv) = open_fetch(&reader_client, "ns-1", &digest_hex, None)
        .await
        .expect("the read-granted identity fetches");
    assert_eq!(ack["source"], json!("store"));
    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    assert_eq!(received, seed, "the read-granted fetch is byte-exact");
    assert_eq!(
        Key::from(Digest::derive(&received, received.len() as u64)),
        key,
        "the consumer's whole-blob hash-check passed"
    );
    drop(send);
    assert_eq!(
        error_code(
            &open_denied(
                &reader_client,
                PUT_OP_ID,
                json!({ "namespace": "ns-1", "len": 4 }),
            )
            .await
        ),
        Some("FORBIDDEN"),
        "the read grant is not the write grant: the read-granted identity cannot put"
    );

    let writer_seed: &[u8] = b"the write-granted put";
    let (_channel_id, mut send, recv, reply) = open_put(
        &writer_client,
        json!({ "namespace": "ns-1", "digest": hex::encode(Digest::derive(writer_seed, writer_seed.len() as u64).bytes()), "len": writer_seed.len() }),
    )
    .await;
    send.write_all(writer_seed).await.expect("the bytes stream");
    drop(send);
    drop(recv);
    let out = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the put resolves")
        .expect("the write-granted identity puts");
    assert_eq!(
        out["digest"],
        json!(hex::encode(
            Digest::derive(writer_seed, writer_seed.len() as u64).bytes()
        ))
    );
    assert_eq!(
        error_code(
            &open_denied(
                &writer_client,
                FETCH_OP_ID,
                json!({ "namespace": "ns-1", "digest": digest_hex.clone() }),
            )
            .await
        ),
        Some("FORBIDDEN"),
        "the write grant is not the read grant: the write-granted identity cannot fetch"
    );
    assert_eq!(
        error_code(
            &open_denied(
                &other_client,
                FETCH_OP_ID,
                json!({ "namespace": "ns-1", "digest": digest_hex.clone() }),
            )
            .await
        ),
        Some("FORBIDDEN"),
        "a read-only peer of another namespace cannot fetch this namespace: grants \
         are per-namespace (closed-by-default)"
    );

    let external = await_call(call_client.call_with_payload(call_payload(
        "blobs/delete",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()] }),
        Some("token:stranger"),
    )))
    .await;
    assert_eq!(
        error_code(&external),
        Some("NOT_FOUND"),
        "delete is not callable from the wire (the internal visibility)"
    );
    let (victim_hex, victim_key) = seeded_entry(&store, b"the deleted entry").await;
    let composed = await_call(call_client.call_with_payload(call_payload(
        "blobs/compose-delete",
        json!({ "namespace": "ns-1", "digests": [victim_hex.clone()] }),
        Some("token:stranger"),
    )))
    .await;
    assert_eq!(
        composed.result.unwrap(),
        json!({ "deleted": [victim_hex.clone()] }),
        "delete ran under the internal authority context only"
    );
    assert!(
        !store.has(&victim_key.as_bytes()).await.unwrap_or(true),
        "the internal delete removed the entry"
    );
}

/// The re-routing posture's serving half (ADR-008 §3): a large
/// (pg-lo) entry served through `blobs/fetch` from the storage
/// node's dual-tier fleet store — the standard channel op, no
/// fleet-specific op (the module supports the shape without
/// change). Env-driven, skipping cleanly server-less.
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
#[tokio::test]
async fn the_rerouting_serving_half_serves_a_large_entry_through_the_standard_fetch() {
    let Some(dsn) = fleet_dsn().await else {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
        return;
    };
    let unique = format!(
        "{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let kv_schema = format!("alkblobs_flows_rerouting_kv_{unique}");
    let lo_schema = format!("alkblobs_flows_rerouting_lo_{unique}");
    let opts = |schema: String| crate::PgOpts {
        schema,
        ..crate::PgOpts::default()
    };
    let storage = Arc::new(
        crate::Store::dual_tier_postgres_pglo_fleet(
            &dsn,
            opts(kv_schema.clone()),
            &dsn,
            opts(lo_schema.clone()),
            1024,
            std::time::Duration::ZERO,
            PAIR_TTL,
        )
        .await
        .expect("the consolidation fleet store constructs against the harness server"),
    );
    let ctx = Arc::new(OpsContext::new(Arc::clone(&storage), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let large: Vec<u8> = (0u32..4096).map(|i| (i % 251) as u8).collect();
    let (key, pin) = storage
        .put(
            Some(large.len() as u64),
            std::io::Cursor::new(large.as_slice()),
        )
        .await
        .expect("the large put publishes");
    drop(pin);
    let digest_hex = hex::encode(key.digest().bytes());

    let kv_rows = harness_table_count(&dsn, &kv_schema, "blobs").await;
    let lo_rows = harness_table_count(&dsn, &lo_schema, "lo_entries").await;
    assert_eq!(
        (kv_rows, lo_rows),
        (0, 1),
        "the over-threshold entry landed on the pg-lo tier (kv 0 rows, large 1): \
         the re-routing storage node's dual-tier shape"
    );

    assert_eq!(FETCH_OP_ID, "channels/blobs/sub");
    assert_eq!(BLOBS_CHANNEL_ALPN, "alk/blobs");
    let (channel_id, ack, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None)
        .await
        .expect("the large entry fetches through the standard channel op");
    assert_eq!(
        ack,
        json!({ "channel_id": channel_id, "source": "store", "len": large.len() }),
        "the ack names the store path and the expected whole"
    );
    let mut received = Vec::new();
    tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received))
        .await
        .expect("the stream reads")
        .expect("the stream ends");
    assert_eq!(received, large, "the pg-lo entry served byte-exact");
    assert_eq!(
        Key::from(Digest::derive(&received, received.len() as u64)),
        key,
        "the consumer's whole-blob hash-check passed"
    );
    drop(send);

    tokio::time::sleep(EXPIRY_HORIZON).await;
    storage
        .delete(&key.as_bytes())
        .await
        .expect("the serving node's operator delete past the row's expiry");
    assert!(
        !storage.has(&key.as_bytes()).await.expect("the has probe"),
        "the post-serving cleanup removed the entry"
    );
    assert_eq!(
        harness_table_count(&dsn, &lo_schema, "lo_entries").await,
        0,
        "zero LO residue on the shared harness database (delete is one tx: \
         lo_unlink + companion-row delete)"
    );

    fleet_schema_drop(&dsn, &kv_schema).await;
    fleet_schema_drop(&dsn, &lo_schema).await;
    let _ = (ctx, harness, client);
}

// ----------------------------------------------------------------
// D-2's malformed-payload sweep (review-001 — the op handlers' error
// arms): the wire schema gate over every op, the post-schema handler
// ingress family, the establishers' form checks, the fetch pump's
// silent-return arms, the put's torn-stream settle, and the
// store-core failures the ops error family surfaces.
// ----------------------------------------------------------------

/// A one-shot `BidiStreamSource` for the direct pump invocations: the
/// canned accept result (a pre-wired stream pair, taken once, or the
/// closed shape) answers immediately, bypassing the channel wrapper —
/// the pump's silent-return arms are exercisable without the full
/// wire.
struct PumpSource {
    accept: parking_lot::Mutex<Option<BiStream>>,
}

impl PumpSource {
    fn closed() -> PumpSource {
        PumpSource {
            accept: parking_lot::Mutex::new(None),
        }
    }

    fn with(server: BiStream) -> PumpSource {
        PumpSource {
            accept: parking_lot::Mutex::new(Some(server)),
        }
    }
}

#[async_trait::async_trait]
impl BidiStreamSource for PumpSource {
    async fn accept_bi(&self) -> Result<BiStream, StreamError> {
        match self.accept.lock().take() {
            Some(stream) => Ok(stream),
            None => Err(StreamError::StreamClosed),
        }
    }

    async fn open_bi(&self) -> Result<BiStream, StreamError> {
        Err(StreamError::StreamClosed)
    }

    fn remote_addr(&self) -> Option<std::net::SocketAddr> {
        None
    }

    fn close(&self, _code: u32, _reason: &str) {}
}

fn pump_conn(server: BiStream) -> Connection {
    Connection::from_source(
        PumpSource::with(server),
        BLOBS_CHANNEL_ALPN.as_bytes().to_vec(),
    )
}

/// A dummy [`Connection`] for direct handler invocations whose input
/// the handler rejects before any data-plane touch (an unconnected end
/// never matters — the guard rejects first).
fn put_conn() -> Connection {
    let (_, server_end) = tokio::io::duplex(4096);
    Connection::from_bidi(server_end, BLOBS_CHANNEL_ALPN.as_bytes().to_vec(), None)
}

/// A pre-wired duplex pair for direct pump invocations: the server
/// half rides into the pump's [`Connection`], the client end stays
/// with the test — dropping the whole client end tears the pump's
/// writes with `BrokenPipe` immediately (the mid-service
/// silent-return shapes; a half-dropped end would only fill the
/// buffer, so the whole end goes).
fn pump_stream_pair() -> (BiStream, tokio::io::DuplexStream) {
    let (client_end, server_end) = tokio::io::duplex(1024 * 1024);
    let (server_read, server_write) = tokio::io::split(server_end);
    (BiStream::from_joined(server_read, server_write), client_end)
}

#[cfg(feature = "mem")]
#[cfg(unix)]
fn chmod_000(path: &std::path::Path) {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o000)).expect("chmod 000");
}

#[cfg(feature = "mem")]
#[cfg(unix)]
fn chmod_755(path: &std::path::Path) {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).expect("chmod 755");
}

/// The wire schema gate is the first rejection for every op: each of
/// the five ops driven with malformed/missing/over-bound payloads
/// answers `INVALID_INPUT` before any handler or store contact (the
/// sweep over `blobs/stat`, `blobs/have`, `blobs/fetch`, `blobs/put`,
/// and `blobs/delete` — whose external discovery is the NOT_FOUND
/// visibility form and whose internal composition still validates).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_malformed_payload_wire_sweep_rejects_every_op_at_the_schema_gate() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble_registrations(vec![
        OpsHarnessRegistration::pair(stat_pair(Arc::clone(&ctx))),
        OpsHarnessRegistration::pair(have_pair(Arc::clone(&ctx))),
        OpsHarnessRegistration::pair(delete_pair(Arc::clone(&ctx)))
            .with_composition_authority(CompositionAuthority::new("blob-operator", [])),
        OpsHarnessRegistration::pair(composer_pair())
            .with_scoped_env(ScopedPeerEnv::new(["blobs/delete"])),
    ]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let call_client = harness.spawn_call_halves();
    let channel_client = harness
        .spawn_channel_halves(
            vec![
                put_openable(Arc::clone(&ctx)),
                fetch_openable(Arc::clone(&ctx)),
            ],
            Some(identity_named("alice")),
        )
        .await;

    let digest_hex = hex::encode(Digest::derive(b"any-payload", 11).bytes());

    let mut invalids: Vec<(&str, Value)> = Vec::new();
    for payload in [
        json!({ "digest": digest_hex.clone() }),
        json!({ "namespace": "ns-1" }),
        json!({ "namespace": "ns-1", "digest": 42 }),
        json!({ "namespace": "ns-1", "digest": "G".repeat(64) }),
        json!({ "namespace": "ns-1", "digest": "a".repeat(65) }),
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "extra": true }),
    ] {
        invalids.push(("blobs/stat", payload));
    }
    for payload in [
        json!({ "namespace": "ns-1" }),
        json!({ "namespace": "ns-1", "digests": digest_hex.clone() }),
        json!({ "namespace": "ns-1", "digests": [42] }),
        json!({ "namespace": "ns-1", "digests": ["G".repeat(64)] }),
        json!({ "namespace": "ns-1", "digests": [], "token": "x".repeat(64) }),
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone(), digest_hex.clone()], "token": "x".repeat(64) }),
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "unexpected": 1 }),
        // S-5's token field bound: the renewal form's `token` carries
        // `maxLength` + the hex pattern — an over-length or non-hex
        // token is the schema gate's `INVALID_INPUT`, never the
        // handler's ingress (whose decode stays bounded by the 64-char
        // length-first guard).
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": "a".repeat(65) }),
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": "G".repeat(64) }),
    ] {
        invalids.push(("blobs/have", payload));
    }
    for payload in [
        json!({ "namespace": "" }),
        json!({ "namespace": "ns-1", "digests": "no-array" }),
        json!({ "namespace": "ns-1", "digests": [42] }),
        json!({ "namespace": "ns-1", "digests": ["G".repeat(64)] }),
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "surprise": 1 }),
    ] {
        invalids.push(("blobs/compose-delete", payload));
    }

    for (op, payload) in invalids {
        let answered = await_call(call_client.call_with_payload(call_payload(
            op,
            payload.clone(),
            Some("token:alice"),
        )))
        .await;
        assert_eq!(
            error_code(&answered),
            Some("INVALID_INPUT"),
            "{op} rejected the malformed payload at the schema gate: {payload} -> {answered:?}"
        );
    }

    // The op-entry order: the ACL verdict denies the empty namespace
    // before the schema gate answers — closed-by-default runs first
    // (the same op with a well-formed but ungranted namespace).
    let acl_first = await_call(call_client.call_with_payload(call_payload(
        "blobs/stat",
        json!({ "namespace": "", "digest": digest_hex.clone() }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        error_code(&acl_first),
        Some("FORBIDDEN"),
        "the op-entry ACL denies the ungranted namespace ahead of the schema gate"
    );

    for payload in [
        json!({ "namespace": "ns-1" }),
        json!({ "namespace": "ns-1", "digest": "G".repeat(64) }),
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "ranges": "no-array" }),
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "ranges": [] }),
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "ranges": [{ "offset": 0 }] }),
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "ranges": [{ "offset": -1, "len": 4 }] }),
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "ranges": [{ "offset": 0, "len": 4, "surprise": 1 }] }),
        // S-2's count bound: the schema's `maxItems` rides the context's
        // `fetch_max_ranges` knob (the default 256) — an over-count open
        // is the registry's `INVALID_INPUT`, never a handler.
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "ranges": std::iter::repeat_n(json!({ "offset": 0, "len": 1 }), 257).collect::<Vec<_>>() }),
    ] {
        let answered = open_denied(&channel_client, FETCH_OP_ID, payload.clone()).await;
        assert_eq!(
            error_code(&answered),
            Some("INVALID_INPUT"),
            "fetch rejected the malformed open payload: {payload}"
        );
    }

    for payload in [
        json!({ "namespace": "ns-1", "digest": "G".repeat(64), "len": 4 }),
        json!({ "namespace": "ns-1", "len": "four" }),
        json!({ "namespace": "ns-1", "len": -1 }),
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": 4, "surprise": 1 }),
    ] {
        let answered = open_denied(&channel_client, PUT_OP_ID, payload.clone()).await;
        assert_eq!(
            error_code(&answered),
            Some("INVALID_INPUT"),
            "put rejected the malformed offer: {payload}"
        );
    }

    let external = await_call(call_client.call_with_payload(call_payload(
        "blobs/delete",
        json!({ "digests": "not even shaped" }),
        Some("token:alice"),
    )))
    .await;
    assert_eq!(
        error_code(&external),
        Some("NOT_FOUND"),
        "the internal visibility verdict precedes validation on the wire"
    );
}

/// The handlers' post-schema ingress family (the "unreachable
/// post-validation" guards are load-bearing for direct and composed
/// calls that bypass the wire): a handler invoked with a schema-passing
/// shape the ingress cannot parse answers the wire-invalid-input form;
/// a malformed digest field answers the declared `INVALID_DIGEST`
/// rejects-never-guesses ingress.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_handler_ingress_sweep_rejects_the_post_schema_shapes() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));

    let (_, stat_handler) = stat_pair(Arc::clone(&ctx));
    let (_, have_handler) = have_pair(Arc::clone(&ctx));
    let (_, delete_handler) = delete_pair(Arc::clone(&ctx));

    for (handler, payload, expected) in [
        (
            &stat_handler,
            json!({}),
            "input failed input_schema validation",
        ),
        (
            &stat_handler,
            json!({ "namespace": "ns-1", "digest": "ZZZ-not-hex" }),
            "the digest is not canonical digest hex",
        ),
        (
            &have_handler,
            json!({}),
            "input failed input_schema validation",
        ),
        (
            &have_handler,
            json!({ "namespace": "ns-1", "digests": ["ZZZ-not-hex"] }),
            "the digest is not canonical digest hex",
        ),
        (
            &delete_handler,
            json!({}),
            "input failed input_schema validation",
        ),
        (
            &delete_handler,
            json!({ "namespace": "ns-1", "digests": ["ZZZ-not-hex"] }),
            "the digest is not canonical digest hex",
        ),
    ] {
        let answered = (handler)(payload.clone(), bare_context()).await;
        match answered.result {
            Err(error) => {
                assert!(
                    error.code == "INVALID_INPUT" || error.code == INVALID_DIGEST,
                    "the ingress swept {payload}: {error:?}"
                );
                assert!(
                    error.message.contains(expected),
                    "the ingress names its reason against {payload}: {error:?}"
                );
            }
            Ok(out) => panic!("the ingress swept {payload}, got {out:?}"),
        }
    }

    // The store-error family's ingest-side complement: have's pure
    // probe folds engine failures into `present: false` — the probes
    // stay probes (ADR-008's discipline) — asserted here against a
    // probe the store never answers (the keys are absent either way).
    let (_, handler) = have_pair(Arc::clone(&ctx));
    let probed = (handler)(
        json!({ "namespace": "ns-1", "digests": [hex::encode(Digest::derive(b"never", 5).bytes())] }),
        bare_context(),
    )
    .await;
    assert_eq!(
        probed.result.unwrap(),
        json!({ "present": [false] }),
        "the pure probe's fold-by-design form"
    );
}

/// The store-core failures are the ops error family's internal form:
/// a large tier whose tree cannot answer (the local tier's committed
/// shard made unreadable — real media produces exactly this failure
/// class) surfaces through the wire as `INTERNAL`, naming the stage —
/// never as a `None`/`false`/missing shape (the probes' error
/// discipline is distinct from the probes' miss discipline).
/// Establishment-side the same failure refuses with reason
/// `handler_error` — never serve unverifiable bytes. Review-001 S-4:
/// every message is the stable domain text — no store-core `Display`
/// detail (path, engine form, os-error fragment) rides the wire on
/// any driven arm. (The have op is excluded by construction: its
/// pure probe folds engine failures to `present: false` —
/// ADR-008's probes-stay-probes discipline — so a store failure
/// there never synthesizes a wire message.)
#[cfg(feature = "mem")]
#[cfg(unix)]
#[tokio::test]
async fn the_store_cores_failures_surface_through_the_ops_error_family() {
    let unique = format!(
        "alkblobs-ops-d2-broken-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let root = std::env::temp_dir().join(&unique);
    std::fs::create_dir_all(&root).expect("the broken-tree root");
    let store = Arc::new(
        Store::dual_tier_mem_local(&root, 8)
            .await
            .expect("the mem+local dual tier constructs"),
    );
    let content: Vec<u8> = (0..64u32).map(|i| (i % 173) as u8).collect();
    let (key, _pin) = store
        .put(
            Some(content.len() as u64),
            std::io::Cursor::new(content.as_slice()),
        )
        .await
        .expect("the large put lands on the local tier");
    let key_hex = hex::encode(key.as_bytes());
    let shard2 = root.join(&key_hex[..2]).join(&key_hex[2..4]);
    chmod_000(&shard2);

    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble_registrations(vec![
        OpsHarnessRegistration::pair(stat_pair(Arc::clone(&ctx))),
        OpsHarnessRegistration::pair(delete_pair(Arc::clone(&ctx)))
            .with_composition_authority(CompositionAuthority::new("blob-operator", [])),
        OpsHarnessRegistration::pair(composer_pair())
            .with_scoped_env(ScopedPeerEnv::new(["blobs/delete"])),
    ]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read", "write"]);
    let call_client = harness.spawn_call_halves();
    let channel_client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;
    let digest_hex = hex::encode(key.digest().bytes());

    let stat = await_call(call_client.call_with_payload(call_payload(
        "blobs/stat",
        json!({ "namespace": "ns-1", "digest": digest_hex.clone() }),
        Some("token:alice"),
    )))
    .await;
    match stat.result {
        Err(error) => {
            assert_eq!(error.code, "INTERNAL", "{error:?}");
            assert_eq!(
                error.message, "the store core failed at stat",
                "the stable domain text: the stage is named, the store core's Display \
                 detail is not (review-001 S-4): {error:?}"
            );
            assert!(
                !error.message.contains('/')
                    && !error.message.contains("io:")
                    && !error.message.contains("Permission"),
                "no path, engine-form, or os-error fragment rides the wire: {error:?}"
            );
        }
        Ok(out) => panic!("the unreadable shard surfaced a stat answer: {out:?}"),
    }

    let deleted = await_call(call_client.call_with_payload(call_payload(
        "blobs/compose-delete",
        json!({ "namespace": "ns-1", "digests": [digest_hex.clone()] }),
        Some("token:alice"),
    )))
    .await;
    match deleted.result {
        Err(error) => {
            assert_eq!(error.code, "INTERNAL", "{error:?}");
            assert_eq!(
                error.message, "the store core failed at delete",
                "the stable domain text: the stage is named, the store core's Display \
                 detail is not (review-001 S-4): {error:?}"
            );
            assert!(
                !error.message.contains('/')
                    && !error.message.contains("io:")
                    && !error.message.contains("Permission"),
                "no path, engine-form, or os-error fragment rides the wire: {error:?}"
            );
        }
        Ok(out) => panic!("the unreadable shard surfaced a delete answer: {out:?}"),
    }

    for (input, stage) in [
        (
            json!({ "namespace": "ns-1", "digest": digest_hex.clone() }),
            "the verified read",
        ),
        (
            json!({ "namespace": "ns-1", "digest": digest_hex.clone(),
                    "ranges": [{ "offset": 0, "len": 4 }] }),
            "the ranged read",
        ),
    ] {
        let refused = tokio::time::timeout(
            TEST_TIMEOUT,
            channel_client.call_open_op(FETCH_OP_ID, input.clone()),
        )
        .await
        .expect("the refused open resolves");
        match refused.result {
            Err(error) => {
                assert_eq!(error.code, "channel:open_failed", "{error:?}");
                let handler_message = error.message.clone();
                assert!(
                    handler_message.contains(stage),
                    "the failed consult names its stage: {error:?} (wanted {stage})"
                );
                assert!(
                    handler_message.contains("the store core failed at")
                        && !handler_message.contains('/')
                        && !handler_message.contains("io:")
                        && !handler_message.contains("Permission"),
                    "the stable domain text: no store-core detail rides the refusal \
                     (review-001 S-4): {error:?}"
                );
            }
            Ok(open) => panic!("the unreadable shard served an open: {open:?} ({stage})"),
        }
    }

    chmod_755(&shard2);
    let _ = std::fs::remove_dir_all(&root);
    let _ = (store, ctx, harness, call_client, channel_client, _pin);
}

/// The establishers' post-schema form checks (the ingest guards the
/// wire gate makes unreachable, driven directly): a malformed digest
/// claim or range shape refuses establishment as a `HandlerError`
/// naming the field — rejects, never guesses, even where the wire
/// schema would have rejected first.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_establisher_ingress_rejects_the_post_schema_offer_shapes() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let auth = super::harness::auth_for(None, BLOBS_CHANNEL_ALPN.as_bytes());

    let (_, put_establisher, _) = put_openables(Arc::clone(&ctx));
    let (_, fetch_establisher, _) = fetch_openables(Arc::clone(&ctx));

    let put_refused = match put_establisher(json!({ "digest": "ZZZ-not-hex" }), auth.clone()).await
    {
        Err(refused) => refused,
        Ok(_) => panic!("the put establisher's digest ingress did not refuse"),
    };
    assert!(
        matches!(put_refused, alkcall::channels::operations::EstablishmentError::HandlerError { ref message }
            if message.contains("the digest failed its ingress")),
        "the put establisher's digest ingress: {put_refused:?}"
    );

    for input in [
        json!({ "digest": "ZZZ-not-hex" }),
        json!({ "digest": hex::encode(Digest::derive(b"ok-digest", 11).bytes()), "ranges": "not-an-array" }),
        json!({ "digest": hex::encode(Digest::derive(b"ok-digest", 11).bytes()), "ranges": [{ "offset": 0 }] }),
    ] {
        let refused = match fetch_establisher(input.clone(), auth.clone()).await {
            Err(refused) => refused,
            Ok(_) => panic!("the fetch establisher's ingest guards swept {input}"),
        };
        assert!(
            matches!(
                refused,
                alkcall::channels::operations::EstablishmentError::HandlerError { .. }
            ),
            "the fetch establisher's ingest guards swept {input}: {refused:?}"
        );
    }

    // S-2's total-bytes bound at the default config (64 MiB): the
    // stable domain text names the bound and the knob's value — no
    // raw payload value rides the message.
    let over_bytes = match fetch_establisher(
        json!({
            "digest": hex::encode(Digest::derive(b"ok-digest", 11).bytes()),
            "ranges": [{ "offset": 0, "len": u64::MAX }, { "offset": 0, "len": u64::MAX }]
        }),
        auth.clone(),
    )
    .await
    {
        Err(refused) => refused,
        Ok(_) => panic!("the over-bytes request was not refused at the ingress"),
    };
    assert!(
        matches!(
            &over_bytes,
            alkcall::channels::operations::EstablishmentError::HandlerError { message }
                if message.contains("exceeds the total-bytes bound")
                    && (message.contains("the limit is 67108864 bytes")
                        || message.contains("the limit of 67108864 bytes"))
                    && !message.contains("offset")
        ),
        "the over-bytes refusal is the stable text at the default bound: {over_bytes:?}"
    );
}

/// The fetch pump's silent-return arms (R-02's lifetime discipline —
/// the handler task tracks the data plane): a pump whose channel never
/// opened, whose plan fails to downcast or unwrap, or whose stream
/// dies mid-service returns without panic — the three named arms plus
/// the mid-service write-failure forms per serving path, each ending
/// the pump alone.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_fetch_pump_silent_return_arms_end_the_pump_without_serving() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let auth = super::harness::auth_for(None, BLOBS_CHANNEL_ALPN.as_bytes());
    let hex_of = |seed: &[u8]| hex::encode(Digest::derive(seed, seed.len() as u64).bytes());

    // The channel never opened: the pump accepts nothing.
    let (_, _, pump) = fetch_openables(Arc::clone(&ctx));
    let conn =
        Connection::from_source(PumpSource::closed(), BLOBS_CHANNEL_ALPN.as_bytes().to_vec());
    let closed = (pump)(json!({}), None, conn, auth.clone());
    tokio::time::timeout(TEST_TIMEOUT, closed)
        .await
        .expect("the never-opened pump resolved")
        .expect("the never-opened pump returned clean");

    // A plan of the wrong type: the downcast refuses — the pump owns
    // its plan type and nothing else serves through it.
    let (_, _, pump) = fetch_openables(Arc::clone(&ctx));
    let (server, _client) = pump_stream_pair();
    let conn = pump_conn(server);
    let downcast_refused = (pump)(
        json!({}),
        Some(Arc::new(0u8) as alkcall::channels::operations::ChannelPlan),
        conn,
        auth.clone(),
    );
    tokio::time::timeout(TEST_TIMEOUT, downcast_refused)
        .await
        .expect("the misplanned pump resolved")
        .expect("the misplanned pump returned clean");

    let big: Vec<u8> = (0..256u32 * 4096).map(|i| (i % 251) as u8).collect();
    let (key, _pin) = store
        .put(Some(big.len() as u64), std::io::Cursor::new(big.as_slice()))
        .await
        .expect("the big entry put");
    let digest_hex = hex_of(&big);

    // A whole-form plan held by another owner: the unwrap refuses —
    // the plan is the establisher's one-shot hand-over, never shared.
    let (_, _, pump) = fetch_openables(Arc::clone(&ctx));
    let whole = fetch_establish_whole(&ctx, &digest_hex, &auth).await;
    let plan = whole.plan.expect("the whole form's plan");
    let held = Arc::clone(&plan);
    let (server, _client) = pump_stream_pair();
    let conn = pump_conn(server);
    let unwrap_refused = (pump)(json!({}), Some(plan), conn, auth.clone());
    tokio::time::timeout(TEST_TIMEOUT, unwrap_refused)
        .await
        .expect("the held-plan pump resolved")
        .expect("the held-plan pump returned clean");
    drop(held);

    // The whole form's service dying at the first write (the
    // consumer's read half gone): the pump returns, the stream torn.
    let (_, _, pump) = fetch_openables(Arc::clone(&ctx));
    let whole = fetch_establish_whole(&ctx, &digest_hex, &auth).await;
    let plan = whole.plan.expect("the whole form's plan");
    let (server, client) = pump_stream_pair();
    drop(client);
    let conn = pump_conn(server);
    let torn_whole = (pump)(json!({}), Some(plan), conn, auth.clone());
    tokio::time::timeout(TEST_TIMEOUT, torn_whole)
        .await
        .expect("the torn whole pump resolved")
        .expect("the torn whole pump returned clean");

    // The ranged form: the slices pump against a torn consumer.
    let (_, fetch_establisher, pump) = fetch_openables(Arc::clone(&ctx));
    let ranged = fetch_establisher(
        json!({
            "digest": digest_hex.clone(),
            "ranges": [{ "offset": 0, "len": big.len() as u64 }, { "offset": 4, "len": 8 }]
        }),
        auth.clone(),
    )
    .await
    .expect("the ranged establishment");
    let plan = ranged.plan.expect("the ranged form's plan");
    let (server, client) = pump_stream_pair();
    drop(client);
    let conn = pump_conn(server);
    let torn_ranged = (pump)(json!({}), Some(plan), conn, auth.clone());
    tokio::time::timeout(TEST_TIMEOUT, torn_ranged)
        .await
        .expect("the torn ranged pump resolved")
        .expect("the torn ranged pump returned clean");

    // The fanout form: the tapped arm's feed to a torn consumer.
    let seed: Vec<u8> = (0..64u32).map(|i| (i % 91) as u8).collect();
    let fanout_hex = hex_of(&seed);
    let fanout_key = Key::from(Digest::from_git_oid_hex(&fanout_hex).expect("the derivation"));
    let publisher = ctx
        .fanout()
        .begin(&fanout_key, SlowSubscriberPolicy::Stall)
        .expect("the broadcast begins");
    let (_, fetch_establisher, pump) = fetch_openables(Arc::clone(&ctx));
    let tapped = fetch_establisher(json!({ "digest": fanout_hex.clone() }), auth.clone())
        .await
        .expect("the fanout tap establishment");
    let plan = tapped.plan.expect("the fanout tap's plan");
    let (server, client) = pump_stream_pair();
    drop(client);
    let conn = pump_conn(server);
    let torn_fanout = (pump)(json!({}), Some(plan), conn, auth.clone());
    publisher.broadcast(Arc::new(seed.clone())).await;
    tokio::time::timeout(TEST_TIMEOUT, torn_fanout)
        .await
        .expect("the torn fanout pump resolved")
        .expect("the torn fanout pump returned clean");
    publisher.finish();

    let _ = (key, _pin, digest_hex);
}

/// The whole form's establishment — the real establisher over the
/// big entry, the plan the pump tests hand around.
#[cfg(any(feature = "mem", feature = "sqlite"))]
async fn fetch_establish_whole(
    ctx: &Arc<OpsContext>,
    digest_hex: &str,
    auth: &alkcall::core::auth::AuthContext,
) -> alkcall::channels::operations::Establishment {
    let (_, establisher, _) = fetch_openables(Arc::clone(ctx));
    establisher(json!({ "digest": digest_hex }), auth.clone())
        .await
        .expect("the whole establishment")
}

/// The put handler's ingress guards and the over-fed stream's settle
/// (the failure shapes the wire gate's happy path never teaches): the
/// post-schema namespace/digest guards answer the invalid-input form
/// under direct invocation; and an over-fed known-length offer — the
/// facade's put ending ahead of the tee — settles through the kill
/// path as `VERIFICATION_REFUSED` with zero residue.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_put_handler_ingress_guards_and_the_torn_stream_settle() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let (_, put_establisher, put_handler) = put_openables(Arc::clone(&ctx));
    let auth = super::harness::auth_for(None, BLOBS_CHANNEL_ALPN.as_bytes());

    for input in [
        json!({}),
        json!({ "namespace": "ns-1", "digest": "ZZZ-not-hex" }),
    ] {
        let refused = put_handler(input.clone(), None, put_conn(), auth.clone())
            .await
            .expect_err("the offer's ingest guards refused");
        assert_eq!(
            refused.code, "INVALID_INPUT",
            "the post-schema offer guards swept {input}: {refused:?}"
        );
    }

    put_establisher(json!({ "namespace": "ns-1" }), auth.clone())
        .await
        .expect("the offer's ack rides establishment");

    // The over-fed known-length offer: the facade's put rejects ahead
    // of the stream (received > len), the tee observes the ended put
    // after its next read, and the settle kills the feed — the
    // response is the facade's rejection through the put's failure
    // family, nothing committed. The stream stays open through the
    // resolve: no EOF path can race the kill.
    let oversized: Vec<u8> = (0..2u32 * 1024 * 1024).map(|i| (i % 251) as u8).collect();
    let digest_hex = hex::encode(Digest::derive(b"four bytes", 4).bytes());
    let (_, _harness, client) = put_serving(&store).await;
    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": 4 }),
    )
    .await;
    send.write_all(&oversized)
        .await
        .expect("the over-feed streams");
    let rejected = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the over-fed put resolves")
        .expect_err("the facade's over-read rejection surfaces");
    let failed = rejected.call_error().expect("the wire error");
    assert_eq!(
        failed.code, VERIFICATION_REFUSED,
        "the put's failure settle through the torn-stream family: {failed:?}"
    );
    assert!(
        failed.message.contains("the put of"),
        "the offered digest is named in the refusal prefix: {failed:?}"
    );
    assert!(
        !store
            .has(
                &Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"))
                    .as_bytes()
            )
            .await
            .expect("the has probe"),
        "nothing committed"
    );
    drop(send);
    drop(recv);
}

/// The put refusal's stable text at the commit-verification synthesis
/// point (review-001 S-4): real media failure — the local tier's
/// committed-shard directory pre-made unwritable, so the staged put's
/// commit-rename fails strictly after the stream's EOF (pin-first
/// holds on the facade path; the failure cannot precede the tee's
/// settle) — settles as `PUT_REFUSED` whose message names the stage
/// and the refusal's shape, with **no** store-core `Display` detail
/// (no path, no `io:` form, no os-error fragment) and nothing
/// committed. (The shared helper's other call site — the store arm's
/// `put_failed_response` settle — is the same stable template.)
#[cfg(feature = "mem")]
#[cfg(unix)]
#[tokio::test]
async fn the_put_refusal_answers_the_stable_text_without_store_core_detail() {
    let unique = format!(
        "alkblobs-ops-s4-put-refusal-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos()
    );
    let root = std::env::temp_dir().join(&unique);
    std::fs::create_dir_all(&root).expect("the commit-failure root");
    let store = Arc::new(
        Store::dual_tier_mem_local(&root, 8)
            .await
            .expect("the mem+local dual tier constructs"),
    );

    let content: Vec<u8> = (0..512u32 * 1024).map(|i| (i % 179) as u8).collect();
    let digest_hex = hex::encode(Digest::derive(&content, content.len() as u64).bytes());
    let key_hex = hex::encode(
        Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation")).as_bytes(),
    );
    let shard = root.join(&key_hex[..2]).join(&key_hex[2..4]);
    std::fs::create_dir_all(&shard).expect("the committed shard dir is pre-created");
    chmod_000(&shard);

    let (_, _harness, client) = put_serving(&store).await;
    let (_channel_id, mut send, recv, reply) = open_put(
        &client,
        json!({ "namespace": "ns-1", "digest": digest_hex.clone(), "len": content.len() }),
    )
    .await;
    send.write_all(&content)
        .await
        .expect("the offered stream writes");
    drop(send);
    let refused = tokio::time::timeout(TEST_TIMEOUT, reply.resolve())
        .await
        .expect("the refused put resolves")
        .expect_err("the commit failure surfaces as the put refusal");
    let failed = refused.call_error().expect("the wire error");
    assert_eq!(
        failed.code, PUT_REFUSED,
        "the commit-time landing is the put-time refusal family: {failed:?}"
    );
    assert_eq!(
        failed.message,
        "the put refused at the commit verification: the store core refused the put — \
         nothing committed",
        "the stable domain text: the stage and shape named, the store core's Display \
         detail dropped (review-001 S-4): {failed:?}"
    );
    drop(recv);

    chmod_755(&shard);
    assert!(
        !store
            .has(
                &Key::from(Digest::from_git_oid_hex(&digest_hex).expect("the derivation"))
                    .as_bytes()
            )
            .await
            .expect("the has probe"),
        "nothing committed"
    );
    assert_eq!(
        store.gc().pin_count(&Key::from(
            Digest::from_git_oid_hex(&digest_hex).expect("the derivation")
        )),
        0,
        "the failed commit's pin released with the facade's guard"
    );
    let _ = std::fs::remove_dir_all(&root);
    let _ = (store, _harness, client);
}

/// The fanout registry's begin-None form and its dead-arm pruning
/// (the A3 shape's edges): a broadcast already live refuses the second
/// begin (the incumbent carries the same bytes under CAS); a
/// subscriber arm whose consumer left is pruned at the next broadcast
/// under the Stall policy too — the registry never grows unbounded on
/// the arms.
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fanout_begin_refuses_the_second_broadcast_and_prunes_dead_arms() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let seed = b"the pruned arm's broadcast";
    let key = Key::from(Digest::derive(seed, seed.len() as u64));

    let incumbent = ctx
        .fanout()
        .begin(&key, SlowSubscriberPolicy::Stall)
        .expect("the first broadcast begins");
    assert!(
        ctx.fanout()
            .begin(&key, SlowSubscriberPolicy::Stall)
            .is_none(),
        "a broadcast already live: the begin-None form, store-arm-only continues"
    );

    let stalled = ctx.fanout().subscribe(&key).expect("the arm subscribes");
    drop(stalled);
    incumbent
        .broadcast(Arc::new(b"the chunk the dead arm skipped".to_vec()))
        .await;
    let mut fresh = ctx
        .fanout()
        .subscribe(&key)
        .expect("the live broadcast admits");
    incumbent
        .broadcast(Arc::new(b"the chunk the fresh arm carries".to_vec()))
        .await;
    assert_eq!(
        fresh.recv().await,
        Some(Arc::new(b"the chunk the fresh arm carries".to_vec())),
        "the pruned arm freed the registry: the fresh arm rides"
    );
    incumbent.finish();
    assert_eq!(
        fresh.recv().await,
        None,
        "the broadcast end is the degrade signal (the fresh arm)"
    );
}

/// A ranged fetch of a committed pool's absent digest is the typed
/// miss — the ranged form's consult reaches the store's
/// `Err(Missing)` and answers the same `unknown_resource` refusal the
/// whole form answers (the miss is the need announcement on both
/// shapes).
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn a_ranged_fetch_of_an_absent_digest_is_the_typed_miss() {
    let store = test_store().await;
    let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
    let harness = OpsHarness::assemble(vec![]);
    harness.with_token("token:alice", identity_named("alice"));
    harness.grant("alice", "ns-1", &["read"]);
    let client = harness
        .spawn_channel_halves(
            vec![fetch_openable(Arc::clone(&ctx))],
            Some(identity_named("alice")),
        )
        .await;

    let absent = hex::encode(Digest::derive(b"ranged never landed", 19).bytes());
    let refused = open_fetch(
        &client,
        "ns-1",
        &absent,
        Some(json!([{ "offset": 0, "len": 4 }])),
    )
    .await;
    match refused {
        Err(error) => {
            let call_error = error.call_error().expect("the wire error");
            assert_eq!(
                call_error.code, "channel:open_failed",
                "the ranged miss refuses establishment: {error:?}"
            );
            assert_eq!(
                error.establishment_reason(),
                Some("unknown_resource"),
                "the typed vocabulary for the absent resource, ranged form too"
            );
            assert!(
                call_error.message.contains(&absent),
                "the refusal carries the digest: {}",
                call_error.message
            );
        }
        Ok(_) => panic!("the absent digest refused, got a ranged channel"),
    }
}

/// The no-side-door audit, its two halves named: (1) the heuristic
/// enumeration — a tripwire, not a guarantee — of the ops module's
/// store access from its own sources: the observed method set equals
/// the public facade verbs plus exactly the crate-internal renewal
/// seam (`ops_pin_check`/`ops_pin_renew`); no `Backend` touch, no
/// engine types, nothing around the facade. (2) The mechanical
/// guarantee, asserted below the enumeration: the non-test source
/// never *names* an engine, backend, or driver identifier at all.
/// Recorded for the wave reviews' gates to re-check. (Adapted for the
/// wave-8 split: the enumeration scans the module's files, each cut
/// at its first `#[cfg(test)]` gate.)
#[test]
fn the_no_side_door_audit_enumerates_the_module_store_access_and_finds_only_the_facade() {
    let sources: String = [
        ("mod.rs", include_str!("mod.rs")),
        ("tokens.rs", include_str!("tokens.rs")),
        ("wire.rs", include_str!("wire.rs")),
        ("specs.rs", include_str!("specs.rs")),
        ("fetch.rs", include_str!("fetch.rs")),
        ("put.rs", include_str!("put.rs")),
    ]
    .iter()
    .map(|(name, source)| {
        let test_tree_start = source.find("#[cfg(test)]").unwrap_or(source.len());
        format!("// {name}\n{}", &source[..test_tree_start])
    })
    .collect();
    let code: String = sources
        .lines()
        .filter(|line| {
            let trimmed = line.trim_start();
            !(trimmed.starts_with("///") || trimmed.starts_with("//!") || trimmed.starts_with("//"))
        })
        .collect::<Vec<_>>()
        .join("\n")
        .replace('\n', " ");

    let mut found: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    for (idx, _) in code.match_indices("store") {
        let mut rest = code[idx + "store".len()..].trim_start();
        if let Some(paren) = rest.strip_prefix('(') {
            rest = paren.trim_start();
            let Some(paren_end) = rest.strip_prefix(')') else {
                continue;
            };
            rest = paren_end.trim_start();
        }
        let Some(rest) = rest.strip_prefix('.') else {
            continue;
        };
        let ident: String = rest
            .chars()
            .take_while(|c| *c == '_' || c.is_ascii_alphanumeric())
            .collect();
        if !ident.is_empty() {
            found.insert(ident);
        }
    }
    let expected: std::collections::BTreeSet<String> = [
        "put",
        "get",
        "stat",
        "has",
        "read_range",
        "delete",
        "fleet",
        "gc",
        "ops_pin_check",
        "ops_pin_renew",
    ]
    .into_iter()
    .map(str::to_owned)
    .collect();
    assert_eq!(
        found, expected,
        "the enumeration recorded: the facade verbs plus exactly the crate-internal \
         renewal seam — anything else in the set is a side-door to remove; anything \
         missing is an audit stale to refresh"
    );

    for forbidden in [
        "Backend",
        "PostgresBackend",
        "PgLoBackend",
        "KvEngine",
        "LargeEngine",
        "rusqlite",
        "tokio_postgres",
        "deadpool",
        "crate::postgres",
        "crate::sqlite",
        "crate::pg_lo",
        "crate::mem",
        "crate::local",
    ] {
        assert!(
            !code.contains(forbidden),
            "no side-door: the ops module's non-test source never names {forbidden}"
        );
    }
}