lifeloop/router/

receipts.rs

//! Receipt synthesis, idempotency, and run-scoped sequencing (issue #14).
//!
//! Fills the [`ReceiptEmitter`] seam declared in `src/router/seams.rs`
//! (issue #7) and implements the status-mapping handoff documented on
//! [`super::NegotiatedPlan`] (issue #13).
//!
//! # Boundary
//!
//! Owns:
//! * [`ReceiptContext`] — caller-supplied identifiers Lifeloop cannot
//!   synthesize (client id, harness ids, root receipt id, parent
//!   receipt id, at-epoch clock value);
//! * [`SequenceGenerator`] — a per-`lifeloop_run_id` monotonic counter
//!   that scopes [`crate::LifecycleReceipt::sequence`] within a run.
//!   Reset semantics are defined in the type docs;
//! * [`IdempotencyStore`] (trait + [`InMemoryIdempotencyStore`]) — the
//!   replay-boundary store keyed by
//!   `(client_id, adapter_id, idempotency_key)`. Persistence is the
//!   consumer's concern; the trait exists so a real consumer can plug
//!   in a database without changing this module;
//! * [`LifeloopReceiptEmitter`] — the concrete emitter that consumes a
//!   [`super::NegotiatedPlan`] + [`crate::CallbackResponse`] and
//!   produces a validated, idempotent [`crate::LifecycleReceipt`];
//! * [`ReceiptError`] — typed failure variants for emission
//!   (validation failures, idempotency conflicts, the
//!   `receipt.emitted` rejection).
//!
//! Does **not** own:
//! * failure-class mapping for arbitrary
//!   [`super::validation::RouteError`]s — that is issue #15's
//!   [`super::FailureMapper`] seam. This module reads
//!   [`crate::FailureClass`] / [`crate::RetryClass`] directly when
//!   building a `status=failed` receipt;
//! * receipt persistence beyond the in-memory idempotency cache.
//!   A consumer's durable receipt ledger plugs in via the
//!   [`IdempotencyStore`] trait;
//! * payload body inspection.
//!
//! # Status mapping (mirrors the `NegotiatedPlan` doc and seeds #15)
//!
//! | Source                                                         | Receipt status      | Failure class                                           |
//! |----------------------------------------------------------------|---------------------|---------------------------------------------------------|
//! | `event == receipt.emitted`                                     | (rejected — error)  | n/a                                                     |
//! | `outcome=Unsupported`                                          | `failed`            | `negotiated.failure_class` (≥ `capability_unsupported`) |
//! | `outcome=RequiresOperator`                                     | `failed`            | `operator_required`                                     |
//! | `outcome=Degraded`                                             | `degraded`          | none                                                    |
//! | `outcome=Satisfied` + any `placement_decisions[].Failed`       | `failed`            | first failed decision's `failure_class`                 |
//! | `outcome=Satisfied` + `response.status=Failed`                 | `failed`            | `response.failure_class` (REQUIRED by validation)       |
//! | `outcome=Satisfied` + `response.status=Skipped`                | `skipped`           | none                                                    |
//! | `outcome=Satisfied` + `response.status=Observed`               | `observed`          | none                                                    |
//! | `outcome=Satisfied` + `response.status=Degraded`               | `degraded`          | none                                                    |
//! | `outcome=Satisfied` + `response.status=Delivered`              | `delivered`         | none                                                    |
//!
//! `retry_class` on a `failed` receipt is seeded via
//! [`crate::FailureClass::default_retry`] unless the response or
//! negotiation already provided one.
//!
//! NOTE: The spec status vocabulary is `observed | delivered | skipped
//! | degraded | failed`. The issue brief's "blocked" shorthand for a
//! halted dispatch maps to `failed` per the spec — there is no
//! separate `blocked` variant on the wire.

use std::collections::HashMap;
use std::sync::Mutex;

use crate::{
    CallbackResponse, FailureClass, LifecycleEventKind, LifecycleReceipt, PayloadReceipt,
    PlacementOutcome, ReceiptStatus, RetryClass, SCHEMA_VERSION, ValidationError,
};

use super::negotiation::{NegotiatedPlan, PayloadPlacementDecision};
use super::seams::ReceiptEmitter;

// ===========================================================================
// ReceiptError
// ===========================================================================

/// Failure variants from receipt emission.
///
/// `Conflict` is the spec-named `duplicate_id_conflict`: the same
/// `(client_id, adapter_id, idempotency_key)` tuple was reused with a
/// receipt body that does not match the prior content.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReceiptError {
    /// `receipt.emitted` is a notification event and must not itself
    /// produce a receipt. Rejected at emit time so a misuse cannot
    /// land in the idempotency store.
    ReceiptEmittedNotEmittable,
    /// Same `idempotency_key` reused with different content. Maps
    /// onto the spec's `duplicate_id_conflict` failure class.
    Conflict { idempotency_key: String },
    /// The synthesized receipt failed [`LifecycleReceipt::validate`].
    Invalid(ValidationError),
}

impl std::fmt::Display for ReceiptError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ReceiptEmittedNotEmittable => f.write_str(
                "receipt.emitted is a notification event and does not itself \
                 produce a receipt",
            ),
            Self::Conflict { idempotency_key } => write!(
                f,
                "idempotency_key `{idempotency_key}` was reused with \
                 different receipt content (duplicate_id_conflict)"
            ),
            Self::Invalid(e) => write!(f, "synthesized receipt failed validation: {e}"),
        }
    }
}

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

impl From<ValidationError> for ReceiptError {
    fn from(e: ValidationError) -> Self {
        Self::Invalid(e)
    }
}

// ===========================================================================
// ReceiptContext
// ===========================================================================

/// Caller-supplied identifiers and clock value Lifeloop cannot
/// synthesize on its own. Every field except the harness ids is
/// required.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReceiptContext {
    /// Client-declared stable label scoping idempotency and replay.
    /// Each lifecycle client picks its own opaque identifier; see the
    /// spec's "Receipt Schema" section in
    /// `docs/specs/lifecycle-contract/body.md` for the illustrative
    /// client-label list. Required-non-empty.
    pub client_id: String,
    /// Opaque identifier for *this* emitted receipt. Required-non-empty.
    pub receipt_id: String,
    /// Optional parent receipt id for nested or causally linked
    /// lifecycle receipts. `None` for root receipts.
    pub parent_receipt_id: Option<String>,
    /// Wall-clock value to stamp on the receipt. Required.
    pub at_epoch_s: u64,
    /// Optional harness session id correlation.
    pub harness_session_id: Option<String>,
    /// Optional harness run id correlation. Receipts in the same
    /// `lifeloop_run_id` (= `harness_run_id` when present) share a
    /// monotonic [`SequenceGenerator`] counter; absent run id means
    /// no in-run sequence is synthesized (`sequence` = `None`).
    pub harness_run_id: Option<String>,
    /// Optional harness task id correlation.
    pub harness_task_id: Option<String>,
}

// ===========================================================================
// SequenceGenerator
// ===========================================================================

/// Per-`lifeloop_run_id` monotonic sequence counter.
///
/// The spec ties `sequence` to "the strongest available durable
/// session scope". For a Lifeloop-synthesized receipt, that scope is
/// the `lifeloop_run_id` (= caller's `harness_run_id`). The generator
/// keeps one counter per run id and hands out 1, 2, 3, ... within a
/// run.
///
/// Sequence is per run, *not* global: emitting under run A then run
/// B then run A again produces A=1, B=1, A=2.
///
/// When the caller supplies no `harness_run_id`, the emitter sets
/// `sequence=None` rather than inventing a misleading cross-run order
/// — see the spec's "required and nullable" rule for `sequence`.
#[derive(Debug, Default)]
pub struct SequenceGenerator {
    counters: Mutex<HashMap<String, u64>>,
}

impl SequenceGenerator {
    pub fn new() -> Self {
        Self::default()
    }

    /// Allocate the next sequence value for `run_id` and return it.
    /// Counters start at 1 within a run.
    pub fn next(&self, run_id: &str) -> u64 {
        let mut guard = self.counters.lock().expect("sequence mutex poisoned");
        let slot = guard.entry(run_id.to_string()).or_insert(0);
        *slot += 1;
        *slot
    }
}

// ===========================================================================
// IdempotencyStore
// ===========================================================================

/// Composite key the idempotency store uses internally. The spec
/// scopes idempotency by `(client_id, adapter_id, idempotency_key)`
/// — the key type pins that triple so a consumer's persistent store
/// cannot accidentally collapse the scope.
#[derive(Debug, Clone, Hash, PartialEq, Eq)]
pub struct IdempotencyKey {
    pub client_id: String,
    pub adapter_id: String,
    pub idempotency_key: String,
}

/// Replay-boundary store. Distinct trait so a consumer can plug in a
/// real database without this module knowing.
///
/// "Receipt content" for the replay comparison is the set of
/// idempotency-invariant fields: every field *except* `receipt_id`,
/// `at_epoch_s`, and `sequence`, which are minted fresh on each
/// emission. Implementations MUST compare on that content (the
/// in-memory implementation below is the reference shape) so a
/// spec-compliant retry of the same `idempotency_key` replays rather
/// than conflicting.
pub trait IdempotencyStore {
    /// Look up the prior receipt for `key`, if any.
    fn get(&self, key: &IdempotencyKey) -> Option<LifecycleReceipt>;

    /// Insert or replay-confirm `receipt` under `key`. Implementations
    /// must enforce the spec's idempotency rule (comparing the
    /// idempotency-invariant content defined above, not the
    /// per-emission `receipt_id`/`at_epoch_s`/`sequence`):
    /// * if no prior entry exists → insert and return
    ///   `Ok(StoreOutcome::Inserted)`;
    /// * if a prior entry exists with content equal to `receipt` →
    ///   return `Ok(StoreOutcome::Replayed(prior))` *without*
    ///   overwriting;
    /// * if a prior entry exists with different content → return
    ///   `Err(ReceiptError::Conflict { .. })`.
    fn put(
        &self,
        key: &IdempotencyKey,
        receipt: &LifecycleReceipt,
    ) -> Result<StoreOutcome, ReceiptError>;
}

/// Outcome of a successful [`IdempotencyStore::put`].
///
/// `Replayed` carries a full [`LifecycleReceipt`]. The size disparity
/// between `Inserted` (zero-sized) and `Replayed` triggers
/// `clippy::large_enum_variant`; we accept it rather than boxing
/// because every `put()` returns a stack-local `StoreOutcome` and
/// boxing would force every `IdempotencyStore` impl through an extra
/// indirection for no measurable benefit (mirrors the choice for
/// `AdapterResolution::Found` in `validation.rs`).
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(clippy::large_enum_variant)]
pub enum StoreOutcome {
    /// First write under this key. The supplied receipt is now the
    /// canonical value.
    Inserted,
    /// Replay of an existing receipt; the prior value is returned
    /// unchanged.
    Replayed(LifecycleReceipt),
}

/// In-memory reference [`IdempotencyStore`] backed by a `HashMap`
/// behind a `Mutex` for interior mutability through `&self`.
///
/// Suitable for tests and for single-process consumers who do not
/// need durable replay protection.
#[derive(Debug, Default)]
pub struct InMemoryIdempotencyStore {
    inner: Mutex<HashMap<IdempotencyKey, LifecycleReceipt>>,
}

impl InMemoryIdempotencyStore {
    pub fn new() -> Self {
        Self::default()
    }
}

impl IdempotencyStore for InMemoryIdempotencyStore {
    fn get(&self, key: &IdempotencyKey) -> Option<LifecycleReceipt> {
        self.inner
            .lock()
            .expect("idem mutex poisoned")
            .get(key)
            .cloned()
    }

    fn put(
        &self,
        key: &IdempotencyKey,
        receipt: &LifecycleReceipt,
    ) -> Result<StoreOutcome, ReceiptError> {
        let mut guard = self.inner.lock().expect("idem mutex poisoned");
        if let Some(prior) = guard.get(key) {
            if replay_content_eq(prior, receipt) {
                return Ok(StoreOutcome::Replayed(prior.clone()));
            }
            return Err(ReceiptError::Conflict {
                idempotency_key: key.idempotency_key.clone(),
            });
        }
        guard.insert(key.clone(), receipt.clone());
        Ok(StoreOutcome::Inserted)
    }
}

/// Replay content-equality over idempotency-invariant fields.
///
/// The spec's idempotency rule keys a replay on identical *receipt
/// content*, but `receipt_id`, `at_epoch_s`, and `sequence` are
/// minted fresh on every emission (a new opaque id, a new wall-clock
/// stamp, and a per-run monotonic counter). Comparing the whole
/// receipt would therefore turn a spec-compliant retry of the same
/// `idempotency_key` into a false `duplicate_id_conflict`. We exclude
/// exactly those per-emission fields and compare the rest, so a retry
/// that produces the same logical receipt replays cleanly.
fn replay_content_eq(a: &LifecycleReceipt, b: &LifecycleReceipt) -> bool {
    a.schema_version == b.schema_version
        && a.idempotency_key == b.idempotency_key
        && a.client_id == b.client_id
        && a.adapter_id == b.adapter_id
        && a.invocation_id == b.invocation_id
        && a.event == b.event
        && a.event_id == b.event_id
        && a.parent_receipt_id == b.parent_receipt_id
        && a.integration_mode == b.integration_mode
        && a.status == b.status
        && a.harness_session_id == b.harness_session_id
        && a.harness_run_id == b.harness_run_id
        && a.harness_task_id == b.harness_task_id
        && a.payload_receipts == b.payload_receipts
        && a.telemetry_summary == b.telemetry_summary
        && a.capability_degradations == b.capability_degradations
        && a.failure_class == b.failure_class
        && a.retry_class == b.retry_class
        && a.warnings == b.warnings
}

// ===========================================================================
// LifeloopReceiptEmitter
// ===========================================================================

/// Concrete [`ReceiptEmitter`] for issue #14.
///
/// Holds a sequence generator and an idempotency store. The
/// [`ReceiptEmitter::emit`] trait method takes a fully-built
/// [`LifecycleReceipt`] (the seam from issue #7) and validates +
/// idempotency-checks it. The richer
/// [`Self::synthesize_and_emit`] entry point consumes a
/// [`NegotiatedPlan`] + [`CallbackResponse`] + [`ReceiptContext`] and
/// builds the receipt before storing it — that is the path issue #14
/// is delivering.
#[derive(Debug)]
pub struct LifeloopReceiptEmitter<S: IdempotencyStore = InMemoryIdempotencyStore> {
    sequencer: SequenceGenerator,
    store: S,
}

impl LifeloopReceiptEmitter<InMemoryIdempotencyStore> {
    /// Construct an emitter backed by an in-memory idempotency store.
    pub fn in_memory() -> Self {
        Self {
            sequencer: SequenceGenerator::new(),
            store: InMemoryIdempotencyStore::new(),
        }
    }
}

impl<S: IdempotencyStore> LifeloopReceiptEmitter<S> {
    /// Construct an emitter with a caller-supplied store (e.g. a
    /// database-backed implementation in production).
    pub fn with_store(store: S) -> Self {
        Self {
            sequencer: SequenceGenerator::new(),
            store,
        }
    }

    pub fn store(&self) -> &S {
        &self.store
    }

    pub fn sequencer(&self) -> &SequenceGenerator {
        &self.sequencer
    }

    /// Build, validate, and idempotency-check a receipt from a
    /// negotiation result + the client's response.
    ///
    /// Returns the canonical receipt (newly inserted, or the prior
    /// replay value when the same idempotency key is reused with the
    /// same content).
    pub fn synthesize_and_emit(
        &self,
        negotiated: &NegotiatedPlan,
        response: &CallbackResponse,
        ctx: &ReceiptContext,
    ) -> Result<LifecycleReceipt, ReceiptError> {
        // `receipt.emitted` is a notification event and never itself
        // produces a receipt. Reject at emit time so the
        // idempotency store cannot record an illegal receipt even
        // before validate() runs.
        if matches!(negotiated.plan.event, LifecycleEventKind::ReceiptEmitted) {
            return Err(ReceiptError::ReceiptEmittedNotEmittable);
        }

        let (status, failure_class, retry_class) = derive_status(negotiated, response);

        // Replay check BEFORE sequence allocation: if this idempotency
        // key has already been recorded, reuse the prior sequence so a
        // content-equality replay short-circuits cleanly. Otherwise a
        // replay would synthesize seq=N+1, fail the content-equality
        // check, and surface as a false Conflict.
        let prior_sequence = negotiated.plan.idempotency_key.as_deref().and_then(|idem| {
            let key = IdempotencyKey {
                client_id: ctx.client_id.clone(),
                adapter_id: negotiated.plan.adapter.adapter_id.clone(),
                idempotency_key: idem.to_string(),
            };
            self.store.get(&key).map(|prior| prior.sequence)
        });

        // Sequence is scoped to lifeloop_run_id (= harness_run_id
        // when supplied). Without a run id, leave sequence null.
        let sequence = match prior_sequence {
            Some(reused) => reused,
            None => ctx
                .harness_run_id
                .as_deref()
                .map(|run| self.sequencer.next(run)),
        };

        let receipt = LifecycleReceipt {
            schema_version: SCHEMA_VERSION.to_string(),
            receipt_id: ctx.receipt_id.clone(),
            idempotency_key: negotiated.plan.idempotency_key.clone(),
            client_id: ctx.client_id.clone(),
            adapter_id: negotiated.plan.adapter.adapter_id.clone(),
            invocation_id: negotiated.plan.invocation_id.clone(),
            event: negotiated.plan.event,
            event_id: negotiated.plan.event_id.clone(),
            sequence,
            parent_receipt_id: ctx.parent_receipt_id.clone(),
            integration_mode: negotiated.plan.integration_mode,
            status,
            at_epoch_s: ctx.at_epoch_s,
            harness_session_id: ctx.harness_session_id.clone(),
            harness_run_id: ctx.harness_run_id.clone(),
            harness_task_id: ctx.harness_task_id.clone(),
            payload_receipts: payload_receipts_from(negotiated),
            telemetry_summary: serde_json::Map::new(),
            capability_degradations: negotiated.capability_degradations.clone(),
            failure_class,
            retry_class,
            warnings: merged_warnings(negotiated, response),
        };

        receipt.validate()?;

        match negotiated.plan.idempotency_key.as_deref() {
            Some(idem) => {
                let key = IdempotencyKey {
                    client_id: ctx.client_id.clone(),
                    adapter_id: negotiated.plan.adapter.adapter_id.clone(),
                    idempotency_key: idem.to_string(),
                };
                match self.store.put(&key, &receipt)? {
                    StoreOutcome::Inserted => Ok(receipt),
                    StoreOutcome::Replayed(prior) => Ok(prior),
                }
            }
            // No idempotency key — the receipt itself is the replay
            // boundary; we hand it back unstored. A future durable
            // ledger may still record it under `receipt_id`, but
            // that is the consumer's concern.
            None => Ok(receipt),
        }
    }
}

impl<S: IdempotencyStore> ReceiptEmitter for LifeloopReceiptEmitter<S> {
    type Error = ReceiptError;

    /// Validate + idempotency-check an externally-built
    /// [`LifecycleReceipt`]. The richer
    /// [`Self::synthesize_and_emit`] is the path issue #14 is
    /// delivering; this trait method exists to satisfy the issue #7
    /// seam contract for callers that already hold a built receipt.
    fn emit(&self, receipt: &LifecycleReceipt) -> Result<(), Self::Error> {
        receipt.validate()?;
        if let Some(idem) = receipt.idempotency_key.as_deref() {
            let key = IdempotencyKey {
                client_id: receipt.client_id.clone(),
                adapter_id: receipt.adapter_id.clone(),
                idempotency_key: idem.to_string(),
            };
            self.store.put(&key, receipt)?;
        }
        Ok(())
    }
}

// ===========================================================================
// Status derivation (the public mapping table mirrored at the top of file)
// ===========================================================================

fn derive_status(
    negotiated: &NegotiatedPlan,
    response: &CallbackResponse,
) -> (ReceiptStatus, Option<FailureClass>, Option<RetryClass>) {
    use crate::NegotiationOutcome as NO;

    // Capability outcomes that block dispatch always win.
    match negotiated.outcome {
        NO::Unsupported => {
            let fc = negotiated
                .failure_class
                .unwrap_or(FailureClass::CapabilityUnsupported);
            return (ReceiptStatus::Failed, Some(fc), Some(fc.default_retry()));
        }
        NO::RequiresOperator => {
            let fc = FailureClass::OperatorRequired;
            return (ReceiptStatus::Failed, Some(fc), Some(fc.default_retry()));
        }
        NO::Degraded => {
            // Fall through to client-status-aware refinement below
            // unless the client itself reported Failed.
            if matches!(response.status, ReceiptStatus::Failed) {
                let fc = response
                    .failure_class
                    .unwrap_or(FailureClass::InternalError);
                let rc = response.retry_class.unwrap_or(fc.default_retry());
                return (ReceiptStatus::Failed, Some(fc), Some(rc));
            }
            return (ReceiptStatus::Degraded, None, None);
        }
        NO::Satisfied => {}
    }

    // Satisfied negotiation: a placement decision may still have
    // failed (e.g. payload_too_large on a non-blocking outcome —
    // negotiation already escalates to Unsupported when this
    // happens, but be defensive).
    if let Some(failed) = negotiated.placement_decisions.iter().find_map(|d| match d {
        PayloadPlacementDecision::Failed { failure_class, .. } => Some(*failure_class),
        _ => None,
    }) {
        return (
            ReceiptStatus::Failed,
            Some(failed),
            Some(failed.default_retry()),
        );
    }

    // Honor the client's reported status.
    match response.status {
        ReceiptStatus::Failed => {
            let fc = response
                .failure_class
                .unwrap_or(FailureClass::InternalError);
            let rc = response.retry_class.unwrap_or(fc.default_retry());
            (ReceiptStatus::Failed, Some(fc), Some(rc))
        }
        ReceiptStatus::Skipped => (ReceiptStatus::Skipped, None, None),
        ReceiptStatus::Observed => (ReceiptStatus::Observed, None, None),
        ReceiptStatus::Degraded => (ReceiptStatus::Degraded, None, None),
        ReceiptStatus::Delivered => (ReceiptStatus::Delivered, None, None),
    }
}

fn payload_receipts_from(negotiated: &NegotiatedPlan) -> Vec<PayloadReceipt> {
    // Map placement decisions onto wire PayloadReceipts. Provenance
    // comes from the payload envelope that negotiation evaluated, not
    // from CallbackResponse::client_payloads; response payloads are
    // client output and may be sparse or derived.
    negotiated
        .placement_decisions
        .iter()
        .map(|d| match d {
            PayloadPlacementDecision::Chosen {
                payload_id,
                payload_kind,
                byte_size,
                content_digest,
                chosen,
                ..
            } => PayloadReceipt {
                payload_id: payload_id.clone(),
                payload_kind: payload_kind.clone(),
                placement: *chosen,
                status: PlacementOutcome::Delivered,
                byte_size: *byte_size,
                content_digest: content_digest.clone(),
            },
            PayloadPlacementDecision::Failed {
                payload_id,
                payload_kind,
                byte_size,
                content_digest,
                rejected,
                ..
            } => PayloadReceipt {
                payload_id: payload_id.clone(),
                payload_kind: payload_kind.clone(),
                // No placement won; surface the first attempted one
                // for diagnostics. Falls back to ReceiptOnly when
                // there were no attempts (unusual).
                placement: rejected
                    .first()
                    .map(|r| r.placement())
                    .unwrap_or(crate::PlacementClass::ReceiptOnly),
                status: PlacementOutcome::Failed,
                byte_size: *byte_size,
                content_digest: content_digest.clone(),
            },
            PayloadPlacementDecision::Skipped {
                payload_id,
                payload_kind,
                byte_size,
                content_digest,
                rejected,
            } => PayloadReceipt {
                payload_id: payload_id.clone(),
                payload_kind: payload_kind.clone(),
                // No placement won, but only preferred/optional
                // placements were unavailable; surface the first
                // attempted placement for diagnostics.
                placement: rejected
                    .first()
                    .map(|r| r.placement())
                    .unwrap_or(crate::PlacementClass::ReceiptOnly),
                status: PlacementOutcome::Skipped,
                byte_size: *byte_size,
                content_digest: content_digest.clone(),
            },
        })
        .collect()
}

fn merged_warnings(
    negotiated: &NegotiatedPlan,
    response: &CallbackResponse,
) -> Vec<crate::Warning> {
    let mut out = negotiated.warnings.clone();
    out.extend(response.warnings.iter().cloned());
    out
}