ff-core 0.3.2

//! Partition families, keys, and routing helpers.
//!
//! # Opaque wire keys (issue #91)
//!
//! Public wire surfaces carry [`PartitionKey`] — an opaque
//! `#[serde(transparent)]` newtype over the hash-tag literal
//! (`"{fp:7}"`) — rather than the internal [`Partition`] struct. The
//! [`PartitionFamily`] enum (including its `Execution` / `Flow`
//! alias) remains a public `ff-core` type — in-tree consumers that
//! construct a [`Partition`] directly still name it — but it is no
//! longer part of the HTTP / SDK wire DTOs, so external consumers
//! never deserialize it off the wire.
//!
//! ## Migration note (0.2 → 0.3)
//!
//! The HTTP response body for `POST /v1/workers/{id}/claim` changed
//! from
//!
//! ```json
//! {
//!   "execution_id": "{fp:7}:...",
//!   "partition_family": "flow",
//!   "partition_index": 7,
//!   "grant_key": "...",
//!   "expires_at_ms": 0
//! }
//! ```
//!
//! to
//!
//! ```json
//! {
//!   "execution_id": "{fp:7}:...",
//!   "partition_key": "{fp:7}",
//!   "grant_key": "...",
//!   "expires_at_ms": 0
//! }
//! ```
//!
//! Consumers that need the parsed family/index call
//! [`PartitionKey::parse`] (or [`crate::contracts::ClaimGrant::partition`] /
//! [`crate::contracts::ReclaimGrant::partition`]) — these return a
//! typed [`Partition`] so routing code stays unchanged.
//!
//! The `Execution` family label collapses to `Flow` on round-trip
//! (both produce `{fp:N}` under RFC-011 §11 co-location). Consumers
//! that read `grant.partition()?.family` for LOGGING will see
//! `Flow` where `Execution` previously appeared — cosmetically
//! visible, routing-equivalent. See [`PartitionKey`] rustdoc for
//! the full contract.

use crate::types::{BudgetId, ExecutionId, FlowId, LaneId, QuotaPolicyId};
use serde::{Deserialize, Serialize};

/// The partition families in FlowFabric.
///
/// Post-RFC-011: `Execution` and `Flow` are now routing **aliases** —
/// both produce the same `{fp:N}` hash-tag, because execution keys
/// co-locate with their parent flow's partition under hash-tag
/// co-location. The `Execution` variant is **deliberately retained**
/// per RFC-011 §11 Non-goals ("Deleting `PartitionFamily::Execution`
/// from the public API. The variant stays for API compatibility; only
/// its routing behaviour changes."), so downstream crates like
/// cairn-fabric that construct `Partition { family: Execution, .. }`
/// continue to compile and route correctly without source changes.
///
/// New FF-internal code should prefer `PartitionFamily::Flow` for
/// clarity — the `Execution` alias exists solely to preserve the
/// public-API contract promised by RFC-011. The logical distinction
/// between exec-scoped and flow-scoped keys continues to live in the
/// key-name prefix (`ff:exec:*` vs `ff:flow:*`), not in the
/// `PartitionFamily` discriminator.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PartitionFamily {
    /// Flow-structural + execution partition: `{fp:N}` — flow topology
    /// and all per-execution keys co-located with their parent flow.
    Flow,
    /// Execution partition — **routing alias for [`PartitionFamily::Flow`]**
    /// under RFC-011 co-location. Produces `{fp:N}` hash-tags identical to
    /// `Flow` and indexes into `num_flow_partitions`. Kept as a distinct
    /// variant per RFC-011 §11 for downstream-API compatibility (cairn-
    /// fabric and other consumers that construct this variant directly).
    Execution,
    /// Budget partition: `{b:M}` — budget definitions and usage.
    Budget,
    /// Quota partition: `{q:K}` — quota policies and sliding windows.
    Quota,
}

impl PartitionFamily {
    /// Hash tag prefix for this family.
    ///
    /// `Flow` and `Execution` are aliases and both return `"fp"` — see
    /// the enum-level rustdoc for the RFC-011 §11 compatibility rationale.
    fn prefix(self) -> &'static str {
        match self {
            Self::Flow | Self::Execution => "fp",
            Self::Budget => "b",
            Self::Quota => "q",
        }
    }
}

/// Partition counts for each family. Fixed at deployment time.
///
/// Post-RFC-011: `num_execution_partitions` is retired. All execution
/// keys route via `num_flow_partitions` (exec + flow share a slot under
/// hash-tag co-location). Default bumped 64 → 256 to preserve today's
/// total keyspace fanout.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PartitionConfig {
    pub num_flow_partitions: u16,
    pub num_budget_partitions: u16,
    pub num_quota_partitions: u16,
}

impl Default for PartitionConfig {
    fn default() -> Self {
        Self {
            num_flow_partitions: 256,
            num_budget_partitions: 32,
            num_quota_partitions: 32,
        }
    }
}

impl PartitionConfig {
    /// Get the partition count for a given family.
    ///
    /// `Flow` and `Execution` return the same value (`num_flow_partitions`)
    /// — they are routing aliases under RFC-011 co-location.
    pub fn count_for(&self, family: PartitionFamily) -> u16 {
        match family {
            PartitionFamily::Flow | PartitionFamily::Execution => self.num_flow_partitions,
            PartitionFamily::Budget => self.num_budget_partitions,
            PartitionFamily::Quota => self.num_quota_partitions,
        }
    }
}

/// A resolved partition within a family.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct Partition {
    pub family: PartitionFamily,
    pub index: u16,
}

impl Partition {
    /// Returns the Valkey hash tag for this partition, e.g. `{fp:7}`, `{b:0}`.
    pub fn hash_tag(&self) -> String {
        format!("{{{prefix}:{index}}}", prefix = self.family.prefix(), index = self.index)
    }
}

impl std::fmt::Display for Partition {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.hash_tag())
    }
}

/// Opaque routing handle for a partition, as exchanged on public wire
/// surfaces.
///
/// `PartitionKey` is the hash-tag literal (`"{fp:7}"`, `"{b:0}"`, ...)
/// wrapped in a `#[serde(transparent)]` newtype. It is the wire form
/// of [`Partition`] on the public API — crates like `ff-server` and
/// `ff-sdk` exchange `PartitionKey`, not [`Partition`], so the
/// internal `PartitionFamily` enum (with its RFC-011 §11 alias
/// between `Flow` and `Execution`) is never exposed on the wire.
///
/// # Opaque-ness contract
///
/// Consumers MUST treat a `PartitionKey` as opaque: pass it back to
/// FlowFabric on subsequent calls, but do NOT parse the interior hash
/// tag to make routing or policy decisions. Compatibility is only
/// guaranteed for opaque round-tripping of keys PRODUCED by
/// FlowFabric — consumers must not hand-construct hash-tag strings
/// nor rely on non-canonical shapes being accepted. FlowFabric may
/// narrow the accepted shape (e.g. hash-tag alphabet, length bounds)
/// in future minor releases for producer-minted keys without a
/// semver bump, because every such key still round-trips under the
/// new rules. Consumers that need the parsed form call
/// [`PartitionKey::parse`] / [`Self::as_partition`], which returns a
/// typed error on malformed input.
///
/// # Round-trip + alias collapse
///
/// `From<&Partition> -> PartitionKey` is infallible (hash-tag
/// construction never fails). `PartitionKey::parse` → `Partition` is
/// fallible (rejects malformed keys). The round trip
/// `Partition -> PartitionKey -> Partition` is **lossy for the
/// `Execution` alias** — a `Partition { family: Execution, .. }`
/// produces `"{fp:N}"`, which parses back to
/// `Partition { family: Flow, .. }`. Routing and key construction are
/// unaffected (both families emit identical hash tags under RFC-011
/// §11), but consumers that read `grant.partition()?.family` for
/// logging will see `Flow` where `Execution` previously appeared.
#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct PartitionKey(String);

impl PartitionKey {
    /// Return the underlying hash-tag literal (`"{fp:7}"`). Intended
    /// for consumers that pass the key through to another FlowFabric
    /// call; NOT for parsing out the partition family/index.
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Parse a hash-tag literal into a typed [`Partition`]. Accepts
    /// the exact shape produced by [`Partition::hash_tag`] (`{fp:N}`,
    /// `{b:N}`, `{q:N}` where `N` is a `u16`).
    pub fn parse(&self) -> Result<Partition, PartitionKeyParseError> {
        Partition::try_from(self)
    }

    /// Convenience alias for [`Self::parse`]. Present so callers
    /// reading a `ClaimGrant.partition()` helper can chain without
    /// naming the conversion explicitly.
    pub fn as_partition(&self) -> Result<Partition, PartitionKeyParseError> {
        self.parse()
    }
}

impl std::fmt::Display for PartitionKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl From<&Partition> for PartitionKey {
    fn from(p: &Partition) -> Self {
        Self(p.hash_tag())
    }
}

impl From<Partition> for PartitionKey {
    fn from(p: Partition) -> Self {
        Self::from(&p)
    }
}

/// Errors produced when parsing a [`PartitionKey`] back into a
/// [`Partition`].
///
/// The key is treated as opaque on the wire; parsing failures surface
/// as `Err` so malformed producer output fails loudly instead of
/// silently routing to a ghost partition.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PartitionKeyParseError {
    /// Key did not start with `{` and end with `}` (e.g. missing braces).
    MissingBraces,
    /// Interior did not split into exactly `<prefix>:<index>`.
    MalformedBody,
    /// `<prefix>` did not match a known family (`fp`, `b`, `q`).
    UnknownFamilyPrefix(String),
    /// `<index>` did not parse as a `u16`.
    InvalidIndex(String),
}

impl std::fmt::Display for PartitionKeyParseError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::MissingBraces => {
                f.write_str("partition key must be wrapped in '{...}'")
            }
            Self::MalformedBody => {
                f.write_str("partition key body must be '<prefix>:<index>'")
            }
            Self::UnknownFamilyPrefix(p) => {
                write!(f, "unknown partition family prefix '{p}' (expected 'fp', 'b', 'q')")
            }
            Self::InvalidIndex(s) => {
                write!(f, "partition index '{s}' is not a valid u16")
            }
        }
    }
}

impl std::error::Error for PartitionKeyParseError {}

impl TryFrom<&PartitionKey> for Partition {
    type Error = PartitionKeyParseError;

    fn try_from(key: &PartitionKey) -> Result<Self, Self::Error> {
        let s = key.as_str();
        let inner = s
            .strip_prefix('{')
            .and_then(|rest| rest.strip_suffix('}'))
            .ok_or(PartitionKeyParseError::MissingBraces)?;
        let (prefix, index_s) = inner
            .split_once(':')
            .ok_or(PartitionKeyParseError::MalformedBody)?;
        let family = match prefix {
            // `fp` collapses to `Flow`; the `Execution` alias is lost
            // on round-trip by design (RFC-011 §11 — both produce the
            // same hash tag, so routing is preserved; only the
            // metadata-layer family label diverges).
            "fp" => PartitionFamily::Flow,
            "b" => PartitionFamily::Budget,
            "q" => PartitionFamily::Quota,
            other => {
                return Err(PartitionKeyParseError::UnknownFamilyPrefix(
                    other.to_owned(),
                ));
            }
        };
        let index: u16 = index_s
            .parse()
            .map_err(|_| PartitionKeyParseError::InvalidIndex(index_s.to_owned()))?;
        Ok(Partition { family, index })
    }
}

/// Compute CRC16-CCITT of the given bytes, same algorithm as Valkey Cluster.
fn crc16_ccitt(bytes: &[u8]) -> u16 {
    crc16::State::<crc16::XMODEM>::calculate(bytes)
}

/// Compute the partition index for a UUID-based entity.
/// Panics if `num_partitions` is 0 — this is a configuration error.
fn partition_for_uuid(uuid_bytes: &[u8; 16], num_partitions: u16) -> u16 {
    assert!(num_partitions > 0, "num_partitions must be > 0 (division by zero)");
    crc16_ccitt(uuid_bytes) % num_partitions
}

/// Compute the partition for an execution ID.
///
/// Post-RFC-011: decodes the hash-tag prefix out of the id string; does NOT
/// re-hash the UUID. The partition index is baked into the id at mint time
/// via [`ExecutionId::for_flow`] / [`ExecutionId::solo`]. The `config` arg
/// is retained for API symmetry with the other partition functions, but is
/// unused — the exec id carries its partition intrinsically.
pub fn execution_partition(eid: &ExecutionId, _config: &PartitionConfig) -> Partition {
    // `Execution` family preserves exec-scoped semantics at the metadata
    // layer (logs, tracing spans, metric labels) even though routing is
    // aliased to `Flow`'s hash-tag prefix under RFC-011 co-location.
    // See PartitionFamily enum rustdoc for the alias contract.
    Partition {
        family: PartitionFamily::Execution,
        index: eid.partition(),
    }
}

/// Compute the partition for a flow ID.
pub fn flow_partition(fid: &FlowId, config: &PartitionConfig) -> Partition {
    Partition {
        family: PartitionFamily::Flow,
        index: partition_for_uuid(fid.as_bytes(), config.num_flow_partitions),
    }
}

/// Strategy for picking a solo execution's partition from its lane id.
///
/// RFC-011 §5.6 defines the birthday-paradox traffic-amplification
/// mitigation: solo execs hash their lane id to a flow partition with
/// crc16, which can collide with a flow's own partition. Operators
/// that hit a persistent collision install a custom strategy at boot
/// time via [`solo_partition_with`] instead of rebuilding `ff-server`.
///
/// The default impl [`Crc16SoloPartitioner`] matches the algorithm
/// used by every other partition family. Replace only when a real
/// collision surfaces via `ff-server admin partition-collisions`.
pub trait SoloPartitioner: Send + Sync {
    /// Return the partition index for a solo execution on the given lane.
    ///
    /// Must return a value in `0..config.num_flow_partitions`.
    /// Must be deterministic — the same `(lane, config)` always produces
    /// the same index (violated determinism → exec keys would route
    /// differently on each mint, breaking all cross-lookup invariants).
    fn partition_for_lane(&self, lane: &LaneId, config: &PartitionConfig) -> u16;
}

/// Default [`SoloPartitioner`]: `crc16_ccitt(lane_utf8) % num_flow_partitions`.
///
/// Matches the hashing used by [`flow_partition`], [`budget_partition`],
/// [`quota_partition`] — same crc16-CCITT algorithm Valkey Cluster uses
/// for slot assignment.
#[derive(Clone, Copy, Debug, Default)]
pub struct Crc16SoloPartitioner;

impl SoloPartitioner for Crc16SoloPartitioner {
    fn partition_for_lane(&self, lane: &LaneId, config: &PartitionConfig) -> u16 {
        assert!(
            config.num_flow_partitions > 0,
            "num_flow_partitions must be > 0 (division by zero)"
        );
        crc16_ccitt(lane.as_str().as_bytes()) % config.num_flow_partitions
    }
}

/// Compute the partition for a solo (flow-less) execution's lane shard.
///
/// Uses the default [`Crc16SoloPartitioner`]. Deployments that hit a
/// traffic-amplification hotspot (per RFC-011 §5.6) should call
/// [`solo_partition_with`] with a custom partitioner instead.
pub fn solo_partition(lane: &LaneId, config: &PartitionConfig) -> Partition {
    solo_partition_with(lane, config, &Crc16SoloPartitioner)
}

/// Compute the partition for a solo execution using a custom
/// [`SoloPartitioner`] strategy.
///
/// The operator-facing escape hatch for RFC-011 §5.6 traffic-amplification
/// collisions. A deployment that observes a collision via
/// `ff-server admin partition-collisions` instantiates an alternate
/// [`SoloPartitioner`] impl and routes solo mint paths through this
/// function. The default [`Crc16SoloPartitioner`] is used by
/// [`solo_partition`] and [`ExecutionId::solo`] — neither signature
/// changes under this extension point.
pub fn solo_partition_with(
    lane: &LaneId,
    config: &PartitionConfig,
    partitioner: &dyn SoloPartitioner,
) -> Partition {
    // Solo execs are execution-scoped — preserve `Execution` family at
    // the metadata layer. Routing is aliased to `Flow`'s `{fp:N}` tag
    // under RFC-011 co-location; see PartitionFamily rustdoc for the
    // alias contract.
    Partition {
        family: PartitionFamily::Execution,
        index: partitioner.partition_for_lane(lane, config),
    }
}

/// Compute the partition for a budget ID.
pub fn budget_partition(bid: &BudgetId, config: &PartitionConfig) -> Partition {
    Partition {
        family: PartitionFamily::Budget,
        index: partition_for_uuid(bid.as_bytes(), config.num_budget_partitions),
    }
}

/// Compute the partition for a quota policy (by scope ID).
pub fn quota_partition(qid: &QuotaPolicyId, config: &PartitionConfig) -> Partition {
    Partition {
        family: PartitionFamily::Quota,
        index: partition_for_uuid(qid.as_bytes(), config.num_quota_partitions),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn partition_hash_tag_format() {
        let p = Partition { family: PartitionFamily::Flow, index: 7 };
        assert_eq!(p.hash_tag(), "{fp:7}");

        let p = Partition { family: PartitionFamily::Execution, index: 7 };
        assert_eq!(p.hash_tag(), "{fp:7}", "Execution must alias Flow (RFC-011 §11)");

        let p = Partition { family: PartitionFamily::Budget, index: 0 };
        assert_eq!(p.hash_tag(), "{b:0}");

        let p = Partition { family: PartitionFamily::Quota, index: 31 };
        assert_eq!(p.hash_tag(), "{q:31}");
    }

    /// Execution and Flow are deliberate routing aliases post-RFC-011 §11.
    /// This test pins the alias contract so a future edit that diverges the
    /// two routes fails loudly rather than silently breaking co-location.
    #[test]
    fn execution_family_aliases_flow() {
        // Same hash-tag at every index.
        for index in [0u16, 1, 7, 42, 255, 65535] {
            let flow = Partition { family: PartitionFamily::Flow, index };
            let exec = Partition { family: PartitionFamily::Execution, index };
            assert_eq!(
                flow.hash_tag(),
                exec.hash_tag(),
                "Flow and Execution must produce identical hash-tags at index {index}"
            );
        }

        // count_for returns the same value.
        let config = PartitionConfig::default();
        assert_eq!(
            config.count_for(PartitionFamily::Flow),
            config.count_for(PartitionFamily::Execution),
            "count_for(Flow) == count_for(Execution) — both route via num_flow_partitions"
        );

        // A Partition minted with family=Execution produces the same
        // hash-tag as one minted with family=Flow, given the same index.
        // This is the key property cairn-fabric (and any other consumer
        // that constructs `Partition { family: Execution, .. }`) depends on.
        let p_exec = Partition { family: PartitionFamily::Execution, index: 42 };
        let p_flow = Partition { family: PartitionFamily::Flow, index: 42 };
        assert_eq!(p_exec.hash_tag(), p_flow.hash_tag());
        assert_eq!(p_exec.hash_tag(), "{fp:42}");
    }

    #[test]
    fn all_families_produce_distinct_tags() {
        // Post-RFC-011: Flow and Execution deliberately share `{fp:N}` — see
        // `execution_family_aliases_flow`. The three *distinct* tag spaces are
        // {fp:N}, {b:N}, {q:N}. This test asserts that alias-after-collapse.
        let tags: Vec<String> = [
            PartitionFamily::Flow,
            PartitionFamily::Budget,
            PartitionFamily::Quota,
        ]
        .iter()
        .map(|f| Partition { family: *f, index: 0 }.hash_tag())
        .collect();
        let unique: std::collections::HashSet<&String> = tags.iter().collect();
        assert_eq!(unique.len(), 3, "flow/budget/quota must produce distinct hash tags");
    }

    #[test]
    fn flow_partition_determinism() {
        let config = PartitionConfig::default();
        let fid = FlowId::new();
        let p1 = flow_partition(&fid, &config);
        let p2 = flow_partition(&fid, &config);
        assert_eq!(p1, p2);
        assert_eq!(p1.family, PartitionFamily::Flow);
        assert!(p1.index < config.num_flow_partitions);
    }

    #[test]
    fn budget_partition_determinism() {
        let config = PartitionConfig::default();
        let bid = BudgetId::new();
        let p1 = budget_partition(&bid, &config);
        let p2 = budget_partition(&bid, &config);
        assert_eq!(p1, p2);
        assert_eq!(p1.family, PartitionFamily::Budget);
        assert!(p1.index < config.num_budget_partitions);
    }

    #[test]
    fn default_config_values() {
        let config = PartitionConfig::default();
        assert_eq!(config.num_flow_partitions, 256);
        assert_eq!(config.num_budget_partitions, 32);
        assert_eq!(config.num_quota_partitions, 32);
    }

    // ── RFC-011 phase-1 tests ──

    #[test]
    fn execution_id_for_flow_determinism() {
        let config = PartitionConfig::default();
        let fid = FlowId::new();
        let a = ExecutionId::for_flow(&fid, &config);
        let b = ExecutionId::for_flow(&fid, &config);
        // Same flow → same partition (UUID suffix differs).
        assert_eq!(a.partition(), b.partition());
    }

    #[test]
    fn execution_id_solo_determinism() {
        let config = PartitionConfig::default();
        let lane = LaneId::new("workers-a");
        let a = ExecutionId::solo(&lane, &config);
        let b = ExecutionId::solo(&lane, &config);
        // Same lane → same partition.
        assert_eq!(a.partition(), b.partition());
    }

    #[test]
    fn execution_id_partition_matches_flow_partition() {
        let config = PartitionConfig::default();
        let fid = FlowId::new();
        let eid = ExecutionId::for_flow(&fid, &config);
        let fp = flow_partition(&fid, &config);
        assert_eq!(eid.partition(), fp.index);
        let ep = execution_partition(&eid, &config);
        // Indices match (RFC-011 co-location) but family is Execution —
        // preserves exec-scoped semantics at the metadata layer.
        assert_eq!(ep.index, fp.index);
        assert_eq!(ep.family, PartitionFamily::Execution);
        // Alias contract: Execution and Flow produce identical hash-tags.
        assert_eq!(ep.hash_tag(), fp.hash_tag());
    }

    #[test]
    fn execution_partition_reads_hash_tag_not_uuid() {
        // Construct an ExecutionId with a KNOWN partition (0) and a UUID
        // whose crc16-partition lands somewhere else. execution_partition
        // must pick the hash-tag (0), not re-hash the UUID.
        let known_uuid = "550e8400-e29b-41d4-a716-446655440000";
        let s = format!("{{fp:0}}:{known_uuid}");
        let eid = ExecutionId::parse(&s).unwrap();
        let config = PartitionConfig::default();
        let p = execution_partition(&eid, &config);
        assert_eq!(p.index, 0, "must read hash-tag, not re-hash UUID");
    }

    #[test]
    fn execution_partition_ignores_config_value() {
        // Safety test per manager ask #7 (edge case 3):
        // An id minted in one config must decode to the same partition
        // in any other config — the partition is baked into the id,
        // not computed from the config.
        let small = PartitionConfig { num_flow_partitions: 4, ..Default::default() };
        let fid = FlowId::new();
        let eid = ExecutionId::for_flow(&fid, &small);
        let minted_partition = eid.partition();

        // Decode via a different config — must match.
        let big = PartitionConfig { num_flow_partitions: 1024, ..Default::default() };
        let p = execution_partition(&eid, &big);
        assert_eq!(
            p.index, minted_partition,
            "hash-tag is authoritative; config value must not change decoding"
        );
    }

    #[test]
    fn execution_id_parse_rejects_bare_uuid() {
        let bare = "550e8400-e29b-41d4-a716-446655440000";
        match ExecutionId::parse(bare) {
            Err(crate::types::ExecutionIdParseError::MissingTag(_)) => {}
            other => panic!("expected MissingTag, got {other:?}"),
        }
    }

    #[test]
    fn execution_id_parse_accepts_wellformed_shape() {
        let s = "{fp:42}:550e8400-e29b-41d4-a716-446655440000";
        let eid = ExecutionId::parse(s).unwrap();
        assert_eq!(eid.partition(), 42);
        assert_eq!(eid.as_str(), s);
    }

    #[test]
    fn execution_id_parse_rejects_bad_partition_index() {
        // Non-integer partition
        match ExecutionId::parse("{fp:xx}:550e8400-e29b-41d4-a716-446655440000") {
            Err(crate::types::ExecutionIdParseError::InvalidPartitionIndex(_)) => {}
            other => panic!("expected InvalidPartitionIndex, got {other:?}"),
        }
        // u16-overflow partition (65536)
        match ExecutionId::parse("{fp:65536}:550e8400-e29b-41d4-a716-446655440000") {
            Err(crate::types::ExecutionIdParseError::InvalidPartitionIndex(_)) => {}
            other => panic!("expected InvalidPartitionIndex for u16 overflow, got {other:?}"),
        }
    }

    #[test]
    fn execution_id_parse_rejects_bad_uuid() {
        match ExecutionId::parse("{fp:0}:not-a-uuid") {
            Err(crate::types::ExecutionIdParseError::InvalidUuid(_)) => {}
            other => panic!("expected InvalidUuid, got {other:?}"),
        }
    }

    #[test]
    fn solo_partition_determinism() {
        let config = PartitionConfig::default();
        let lane = LaneId::new("workers-a");
        let p1 = solo_partition(&lane, &config);
        let p2 = solo_partition(&lane, &config);
        assert_eq!(p1, p2);
        // Solo execs are execution-scoped — family is Execution, routing
        // is aliased to Flow's `{fp:N}` via PartitionFamily's prefix map.
        assert_eq!(p1.family, PartitionFamily::Execution);
        assert!(p1.index < config.num_flow_partitions);
    }

    #[test]
    fn solo_partition_different_lanes_usually_differ() {
        // Not strict — collisions are possible (and §5.6 acknowledges) —
        // but over 100 distinct lane names we expect most to be distinct.
        let config = PartitionConfig::default();
        let mut seen = std::collections::HashSet::new();
        for i in 0..100 {
            let lane = LaneId::new(format!("lane-{i}"));
            let p = solo_partition(&lane, &config);
            seen.insert(p.index);
        }
        // At 256 partitions with 100 inputs, expect >50 distinct.
        assert!(
            seen.len() > 50,
            "solo_partition distribution too narrow: only {} distinct of 100",
            seen.len()
        );
    }

    // ── SoloPartitioner trait (RFC-011 §5.6 mitigation #3) ──

    #[test]
    fn crc16_solo_partitioner_matches_legacy_behavior() {
        // The default Crc16SoloPartitioner must produce the same index as
        // the pre-trait solo_partition() function — otherwise installing
        // the trait under an existing deployment would silently re-route
        // every solo exec.
        let config = PartitionConfig::default();
        let lane = LaneId::new("workers-a");
        let default_idx = Crc16SoloPartitioner.partition_for_lane(&lane, &config);
        let expected = crc16::State::<crc16::XMODEM>::calculate(lane.as_str().as_bytes())
            % config.num_flow_partitions;
        assert_eq!(default_idx, expected);
    }

    #[test]
    fn solo_partition_with_custom_partitioner_routes_through_trait() {
        // Stub partitioner always returns index 0; a custom impl must
        // override the default routing.
        struct AlwaysZero;
        impl SoloPartitioner for AlwaysZero {
            fn partition_for_lane(&self, _lane: &LaneId, _config: &PartitionConfig) -> u16 {
                0
            }
        }
        let config = PartitionConfig::default();
        let lane = LaneId::new("pick-me");
        let p = solo_partition_with(&lane, &config, &AlwaysZero);
        assert_eq!(p.index, 0);
        // Solo execs → Execution family (aliases Flow for routing).
        assert_eq!(p.family, PartitionFamily::Execution);
    }

    #[test]
    fn solo_partition_default_matches_solo_partition_with_crc16() {
        // solo_partition() and solo_partition_with(Crc16SoloPartitioner)
        // must produce identical Partitions — this pins the default impl
        // as the identity override.
        let config = PartitionConfig::default();
        let lane = LaneId::new("workers-b");
        let default = solo_partition(&lane, &config);
        let explicit = solo_partition_with(&lane, &config, &Crc16SoloPartitioner);
        assert_eq!(default, explicit);
    }

    #[test]
    fn execution_id_serde_via_deserialize_validates() {
        // Valid shape deserialises.
        let json = r#""{fp:0}:550e8400-e29b-41d4-a716-446655440000""#;
        let eid: ExecutionId = serde_json::from_str(json).unwrap();
        assert_eq!(eid.partition(), 0);

        // Bare UUID fails Deserialize.
        let bare = r#""550e8400-e29b-41d4-a716-446655440000""#;
        assert!(serde_json::from_str::<ExecutionId>(bare).is_err());
    }

    // ── PartitionKey (issue #91) ──

    #[test]
    fn partition_key_from_partition_matches_hash_tag() {
        for (family, expected_prefix) in [
            (PartitionFamily::Flow, "fp"),
            (PartitionFamily::Execution, "fp"),
            (PartitionFamily::Budget, "b"),
            (PartitionFamily::Quota, "q"),
        ] {
            let p = Partition { family, index: 42 };
            let k: PartitionKey = (&p).into();
            assert_eq!(k.as_str(), &format!("{{{expected_prefix}:42}}"));
            assert_eq!(k.as_str(), p.hash_tag());
        }
    }

    #[test]
    fn partition_key_round_trip_flow_budget_quota() {
        // Exact round-trip for non-alias families.
        for family in [
            PartitionFamily::Flow,
            PartitionFamily::Budget,
            PartitionFamily::Quota,
        ] {
            let p = Partition { family, index: 7 };
            let k = PartitionKey::from(&p);
            let back = k.parse().expect("must parse");
            assert_eq!(back, p, "round-trip must be identity for {family:?}");
        }
    }

    /// RFC-011 §11 alias collapse: `Execution` round-trips to `Flow`.
    /// Routing is preserved (same hash tag, same `count_for`), only
    /// the metadata-layer label differs. This test pins the documented
    /// collapse so consumers can't quietly depend on `Execution`
    /// surviving a wire hop.
    #[test]
    fn partition_key_collapses_execution_to_flow_on_parse() {
        let p_exec = Partition { family: PartitionFamily::Execution, index: 3 };
        let k = PartitionKey::from(&p_exec);
        assert_eq!(k.as_str(), "{fp:3}");
        let back = k.parse().expect("must parse");
        // Alias collapse: index preserved, family normalises to Flow.
        assert_eq!(back.family, PartitionFamily::Flow);
        assert_eq!(back.index, 3);
        // Routing equivalence: both produce identical hash tags.
        assert_eq!(back.hash_tag(), p_exec.hash_tag());
    }

    #[test]
    fn partition_key_serde_transparent() {
        let p = Partition { family: PartitionFamily::Flow, index: 9 };
        let k = PartitionKey::from(&p);
        let json = serde_json::to_string(&k).unwrap();
        assert_eq!(json, r#""{fp:9}""#, "must serialise as a bare string");
        let parsed: PartitionKey = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed, k);
    }

    #[test]
    fn partition_key_parse_rejects_missing_braces() {
        let k = PartitionKey("fp:0".to_owned());
        assert_eq!(k.parse(), Err(PartitionKeyParseError::MissingBraces));
    }

    #[test]
    fn partition_key_parse_rejects_malformed_body() {
        let k = PartitionKey("{fp0}".to_owned());
        assert_eq!(k.parse(), Err(PartitionKeyParseError::MalformedBody));
    }

    #[test]
    fn partition_key_parse_rejects_unknown_prefix() {
        let k = PartitionKey("{zz:0}".to_owned());
        match k.parse() {
            Err(PartitionKeyParseError::UnknownFamilyPrefix(p)) => assert_eq!(p, "zz"),
            other => panic!("expected UnknownFamilyPrefix, got {other:?}"),
        }
    }

    #[test]
    fn partition_key_parse_rejects_invalid_index() {
        // Non-numeric
        let k = PartitionKey("{fp:xx}".to_owned());
        assert!(matches!(
            k.parse(),
            Err(PartitionKeyParseError::InvalidIndex(_))
        ));
        // u16 overflow
        let k = PartitionKey("{fp:65536}".to_owned());
        assert!(matches!(
            k.parse(),
            Err(PartitionKeyParseError::InvalidIndex(_))
        ));
    }
}