mako-redispatch 0.20.0

//! Redispatch-Aktivierung workflow for Redispatch 2.0.
//!
//! **Direction:** ÜNB → VNB → ANB\
//! **Document:** `redispatch_xml::ActivationDocument`
//!   - `A96` (`Ordered`) — ACO, Anweisung (ÜNB → VNB)
//!   - `A41` (`ActivationResponse`) — ACR, Antwort (ANB/VNB → ÜNB)
//!   - `A42` (`TenderReduction`) — AAR, Teilablehnung
//!
//! # Process description
//!
//! 1. ÜNB sends `ActivationDocument` (ACO, A96) with `status = Ordered` to VNB.
//! 2. VNB cascades the ACO to the relevant ANB(s) within the remaining window.
//! 3. ANB confirms with ACR (A41) or partially rejects with AAR (A42).
//! 4. VNB aggregates responses and sends upstream ACR/AAR to ÜNB.
//!
//! # Two clocks, two sources
//!
//! **The `AcknowledgementDocument` is due in 3 minutes** — „unverzüglich, jedoch
//! spätestens 3 Minuten nach Erhalt der Übertragungsdatei"
//! (`AcknowledgementDocument` FB 1.0g). It answers the *transfer file*, reports a
//! syntax result, and a late one „darf nicht zu einer Fristverletzung des
//! eigentlichen Geschäftsvorfalles führen" — so it never fails the activation
//! it carried. See [`crate::fristen::ACK_FRIST`] and
//! [`crate::fristen::ack_regeln`].
//!
//! **The ACR/AAR window is operator-configured.** It was five minutes under
//! BK6-20-060 §6.3, but BK6-23-241 Tenorziffer 4 repealed that decision and the
//! replacement Prozessbeschreibungen (Tenorziffer 7) are not published yet.
//! [`crate::fristen::`Betreiberfristen`::aktivierung_antwort`] carries the value,
//! defaulting to the historical five minutes.
//!
//! Whatever the configured value, this stays a **real-time** constraint: the
//! `makod` deadline scheduler must tick well inside the window, not on the
//! Werktag cadence `GPKE` and `WiM` use.
//!
//! # Clock semantics
//!
//! Both windows are wall-clock and run in **UTC** — the XSD `UtcDateTime`
//! fields carry explicit `Z` offsets. Only the `BilAReM` Werktag obligations
//! (Wetterdaten, `Stammdaten`-Gültigkeit, Planwert-Überführung) follow German
//! local time.
//!
//! # IFTSTA integration
//!
//! Redispatch 2.0 IFTSTA PIDs (confirmed from IFTSTA AHB 2.1 and PID 4.0):
//!
//! | PID   | Perspective | Description |
//! |-------|-------------|-------------|
//! | 21037 | NB (VNB)    | Kommunikationsprozesse Redispatch — Ansicht NB |
//! | 21038 | BTR         | Kommunikationsprozesse Redispatch — Ansicht BTR |
//!
//! These PIDs are registered by [`crate::RedispatchModule`] into the `PidRouter`
//! and route to this workflow via conversation-ID lookup.
//!
//! PIDs 21035 (GPKE Rückmeldung Lieferstelle → `gpke-supplier-change`),
//! 21036 (`WiM` Strom Teil 1, unassigned) and 21040 (`AWH` Sperrprozesse Gas,
//! unassigned) are **not** Redispatch PIDs and must not be registered here.
//! See `site/content/docs/regulatory/pid-reference.md` for the authoritative PID ownership table.
//!
//! # Regulatory basis
//!
//! `BNetzA` **BK6-23-241** (Beschluss 07.05.2026), Anlage `BilAReM` Kap. 6.3
//! (Abrufprozesse) and Kap. 6.4 (Abstimmung der Ausfallarbeit); EDI@Energy
//! *`ActivationDocument`* FB 1.1f and *`AcknowledgementDocument`* FB 1.0g for the
//! wire format and the acknowledgement Frist.

use mako_engine::{
    deadline::Deadline,
    error::WorkflowError,
    ids::DeadlineId,
    workflow::{CommandPayload, EventPayload, Workflow, WorkflowOutput},
};
use serde::{Deserialize, Serialize};

// ── Workflow name ─────────────────────────────────────────────────────────────

/// Stable workflow name — used in `ProcessRegistry` lookups and log output.
pub const WORKFLOW_NAME: &str = "redispatch-aktivierung";

/// Redispatch IFTSTA PIDs (IFTSTA AHB 2.1, PID 4.0):
///
/// | PID   | Perspective | Description                                      |
/// |-------|-------------|--------------------------------------------------|
/// | 21037 | NB (VNB)    | Kommunikationsprozesse Redispatch — Ansicht NB   |
/// | 21038 | BTR         | Kommunikationsprozesse Redispatch — Ansicht BTR  |
///
/// Only these two PIDs belong to Redispatch 2.0.  PIDs 21035, 21036, and
/// 21040 are NOT Redispatch PIDs — they belong to GPKE and AWH Sperrprozesse
/// respectively (see `site/content/docs/regulatory/pid-reference.md`).
pub const IFTSTA_PIDS: &[u32] = &[21_037, 21_038];

/// Redispatch MSCONS PIDs — time-series data delivery for Ausfallarbeit and EEG.
///
/// These MSCONS messages carry Redispatch 2.0 time-series data and are correlated
/// with the Aktivierung process via conversation-ID (CI tag).
///
/// | PID   | Beschreibung                                       |
/// |-------|----------------------------------------------------|
/// | 13021 | Übermittlung von meteorologischen Daten (Ex-post) — BTR → ANB, ANB → anfNB |
/// | 13022 | Redispatch 2.0 Einzelzeitreihe Ausfallarbeit — BTR ↔ NB, anfNB → ANB |
///
/// Source: BDEW *Anwendungsübersicht Prüfidentifikatoren 4.0* (01.04.2026),
/// sheet *Prüf-ID Prozessschritt*, rows whose Prozessbeschreibung is
/// „Kommunikationsprozesse Redispatch".
///
/// ## What was removed and why
///
/// | PID | Belongs to | Was routed here |
/// |----:|------------|-----------------|
/// | 13020 | `mabis-billing` — Ausfallarbeitsüberführungszeitreihe | ✗ |
/// | 13023 | `mabis-billing` — Lieferantenausfallarbeitssummenzeitreihe | ✗ |
/// | 13026 | „Geschäftsprozesse für EEG-Überführungszeitreihen" | ✗ |
///
/// 13020 and 13023 are `MaBiS` Summenzeitreihen with a full Prüfmitteilung/
/// Datenstatus cycle (IFTSTA 21000, 21002–21005). Routing them to the
/// Aktivierung workflow gave them no settlement stream to live in, so the
/// Prüfmitteilung obligation they carry had nowhere to be recorded. 13026
/// belongs to a third process family entirely.
///
/// ## Relationship to `edmd::domain::REDISPATCH_MSCONS_PIDS`
///
/// The two lists answer different questions and are deliberately not equal:
/// this one says which PIDs route to the Redispatch **workflow**, `edmd`'s says
/// which Ausfallarbeit-related PIDs it **stores intervals for** — a superset,
/// because 13020/13023 route to `mabis-billing` and 13026 to a family that is
/// not implemented, yet all of them still belong in the OLAP store.
///
/// The invariant that has to hold is this set ⊆ `edmd`'s. A PID the workflow
/// accepts whose intervals `edmd` drops would leave the process with no data to
/// settle against. `edmd` does not depend on this crate, so nothing checks it
/// mechanically — the two lists are held together by review.
///
/// A cross-crate dependency from this process-engine crate to `edmd`'s
/// data-tier domain would be the wrong direction architecturally, so the two
/// constants stay separate and the xtask check keeps them consistent.
pub const MSCONS_PIDS: &[u32] = &[13_021, 13_022];

/// Redispatch ORDERS PIDs.
///
/// | PID   | Beschreibung                                  | Von → An            |
/// |-------|-----------------------------------------------|---------------------|
/// | 17209 | Anforderung der Ausfallarbeit durch den anfNB  | NB (anfNB) → NB (ANB) |
///
/// The ANB answers it with MSCONS **13022** (Prozessschritt 2), not with an
/// ORDRSP — which is why this family has no ORDRSP codes at all.
///
/// ## What was removed and why
///
/// | PID | Belongs to |
/// |----:|------------|
/// | 17210 | `mabis-anforderung` — Anforderung Lieferantenausfallarbeitsclearingliste |
/// | 17211 | `mabis-profile` — Reklamation Profile bzw. Profilscharen (EBD `E_0100`) |
/// | 19204 | `mabis-anforderung` — Ablehnung Ab-/Bestellung der Aggregationsebene |
/// | 19301 | Herkunftsnachweisregister — Ablehnung der Anforderung (`S_0092`) |
/// | 19302 | Herkunftsnachweisregister — Bestätigung Ende des Abos (`S_0093`) |
///
/// All five carry the `MaBiS` or HKN-R Prozessbeschreibung in the PID overview,
/// not „Kommunikationsprozesse Redispatch". 19301/19302 are the NB ↔ RB HKN-R
/// exchange and have nothing to do with either family.
///
/// Source: BDEW *Anwendungsübersicht Prüfidentifikatoren 4.0* (01.04.2026).
pub const ORDERS_PIDS: &[u32] = &[17_209];

// ── Deadline labels ───────────────────────────────────────────────────────────

/// Deadline label for the ANB's ACR/AAR window.
///
/// Operator-configured — see
/// [`crate::fristen::`Betreiberfristen`::aktivierung_antwort`]. Historically five
/// minutes under BK6-20-060 §6.3, which BK6-23-241 Tenorziffer 4 repealed.
///
/// Register immediately after [`AktivierungEvent::AcoReceived`] is applied.
/// The `makod` Redispatch scheduler must poll at ≤ 30 s intervals.
pub const ACTIVATION_RESPONSE_WINDOW_LABEL: &str = "redispatch-activation-response-window";

/// Deadline label for the 3-minute `AcknowledgementDocument` window
/// ([`crate::fristen::ACK_FRIST`]).
pub const ACK_WINDOW_LABEL: &str = "redispatch-aktivierung-ack-window";

// ── Enumerations ──────────────────────────────────────────────────────────────

/// Response type from ANB (or upstream VNB → ÜNB).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ResponseType {
    /// Full activation confirmed (ACR, A41).
    Confirmed,
    /// Partial rejection — only a fraction of ordered MW available (AAR, A42).
    PartialRejection,
}

/// Abruf variant in the Aufforderungsfall (`Stammdaten` `AbrufartAufforderungsfall`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum Abrufart {
    /// Z01 — Delta: the schedule states the *change* against the planned value.
    Delta,
    /// Z02 — Sollwert: the schedule states the absolute target value.
    Sollwert,
}

/// How an activation is executed — the central Redispatch 2.0 case split
/// (`BilAReM` Kap. 1: Aufforderungsfall vs Duldungsfall).
///
/// - **Aufforderungsfall**: the Einsatzverantwortliche (EIV/BTR) steers the
///   resource itself according to the transmitted schedule. The 5-minute
///   activation-response window applies — the NB *waits* for ACR/AAR.
/// - **Duldungsfall**: the Netzbetreiber steers the resource directly via the
///   technical Steuerkanal (the operator merely tolerates the measure). No
///   counterparty response is awaited, so the 5-minute response deadline
///   does **not** apply; the NB records its own ACR when steering succeeds.
///
/// Downstream, the case decides the §13a `EnWG` settlement path: Duldungsfall
/// Ausfallarbeit is derived from measured vs. reference Lastgang, while the
/// Aufforderungsfall settles against the transmitted schedule.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "fall")]
pub enum Abwicklung {
    /// EIV-steered per schedule; 5-minute response window enforced.
    Aufforderungsfall {
        /// Delta (Z01) or Sollwert (Z02) schedule semantics.
        abrufart: Abrufart,
    },
    /// NB-steered via Steuerkanal; no response window.
    Duldungsfall,
}

impl Abwicklung {
    /// Whether the 5-minute activation-response window applies.
    #[must_use]
    pub fn response_window_applies(self) -> bool {
        matches!(self, Self::Aufforderungsfall { .. })
    }
}

// ── Events ────────────────────────────────────────────────────────────────────

/// Events emitted by the Aktivierung workflow.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", content = "data")]
pub enum AktivierungEvent {
    /// `ActivationDocument` (ACO, A96 Ordered) received.
    AcoReceived {
        /// MRID (UUID) of the ACO document.
        mrid: String,
        /// Aufforderungsfall vs Duldungsfall for this activation.
        abwicklung: Abwicklung,
        /// Ordered MW to redispatch.
        ordered_mw: f64,
        /// Resource object identifier (NDE coding scheme, BDEW resource code).
        resource_id: String,
        /// Time period of the activation (ISO-8601 interval).
        period: String,
        /// GLN of the sender (ÜNB or VNB).
        sender: String,
        /// GLN of the receiver (VNB or ANB).
        receiver: String,
        /// UTC receipt timestamp.
        received_at: String,
    },
    /// ACO cascaded to ANB (VNB role only).
    AcoCascaded {
        /// GLN of the ANB the ACO was dispatched to.
        recipient_anb: String,
        /// MRID of the outbound ACO.
        child_mrid: String,
    },
    /// ACR (A41) sent — full activation confirmed.
    AcrSent {
        /// MRID of the outbound ACR document.
        acr_mrid: String,
        /// MW actually activated.
        activated_mw: f64,
    },
    /// AAR (A42) sent — partial rejection.
    AarSent {
        /// MRID of the outbound AAR document.
        aar_mrid: String,
        /// MW available (less than ordered).
        available_mw: f64,
        /// Reason code for the rejection.
        reason_code: String,
    },
    /// Response from ANB received (VNB role — aggregating before forwarding).
    AnbResponseReceived {
        /// GLN of the responding ANB.
        anb_id: String,
        /// Whether the ANB confirmed or partially rejected.
        response_type: ResponseType,
        /// MRID of the ANB's response document.
        response_mrid: String,
    },
    /// Aggregated response forwarded upstream to ÜNB (VNB role only).
    ResponseForwarded {
        /// MRID of the upstream ACR or AAR.
        upstream_mrid: String,
        /// Final aggregate response type.
        response_type: ResponseType,
    },
    /// An `AcknowledgementDocument` was received for a sent document.
    AckReceived {
        /// MRID of the `AcknowledgementDocument`.
        ack_mrid: String,
        /// The acknowledged document's MRID.
        acknowledged_mrid: String,
        /// A01 accepted / A02 rejected.
        reason_code: String,
    },
    /// A Redispatch MSCONS/ORDERS/ORDRSP market message was received
    /// (audit record; no state transition).
    MarketMessageReceived {
        /// The message's Prüfidentifikator.
        pid: u32,
        /// Sender MP-ID.
        sender: String,
        /// Receiver MP-ID.
        receiver: String,
        /// EDIFACT message reference.
        message_ref: String,
    },
    /// IFTSTA Vollzugsmeldung received (PID 21037 or 21038).
    IftstaReceived {
        /// IFTSTA Prüfidentifikator (21037 = NB view, 21038 = BTR view).
        pid: u32,
        /// GLN of the sender.
        sender: String,
        /// GLN of the receiver.
        receiver: String,
        /// EDIFACT message reference.
        message_ref: String,
    },
    /// A registered deadline expired.
    DeadlineExpired {
        /// Unique ID of the expired deadline.
        deadline_id: DeadlineId,
        /// Label identifying the deadline type.
        label: Box<str>,
    },
}

impl EventPayload for AktivierungEvent {
    fn event_type(&self) -> &'static str {
        match self {
            Self::AcoReceived { .. } => "AktivierungAcoReceived",
            Self::AcoCascaded { .. } => "AktivierungAcoCascaded",
            Self::AcrSent { .. } => "AktivierungAcrSent",
            Self::AarSent { .. } => "AktivierungAarSent",
            Self::AnbResponseReceived { .. } => "AktivierungAnbResponseReceived",
            Self::ResponseForwarded { .. } => "AktivierungResponseForwarded",
            Self::AckReceived { .. } => "AktivierungAckReceived",
            Self::MarketMessageReceived { .. } => "AktivierungMarketMessageReceived",
            Self::IftstaReceived { .. } => "AktivierungIftstaReceived",
            Self::DeadlineExpired { .. } => "AktivierungDeadlineExpired",
        }
    }
}

// ── Domain data ───────────────────────────────────────────────────────────────

/// Core activation data set at `AcoReceived` time.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AcoData {
    /// MRID of the initiating ACO.
    pub mrid: String,
    /// Aufforderungsfall (EIV steers) or Duldungsfall (NB steers).
    pub abwicklung: Abwicklung,
    /// Ordered MW.
    pub ordered_mw: f64,
    /// Resource identifier.
    pub resource_id: String,
    /// Activation time period.
    pub period: String,
    /// Sender GLN.
    pub sender: String,
    /// Receiver GLN.
    pub receiver: String,
    /// Receipt timestamp.
    pub received_at: String,
}

// ── State ─────────────────────────────────────────────────────────────────────

/// Current state of a Redispatch activation process.
///
/// # Lifecycle (ANB role)
///
/// ```text
/// New → ActivationOrdered → Confirmed | PartialRejection
/// ```
///
/// # Lifecycle (VNB role)
///
/// ```text
/// New → ActivationOrdered → DispatchedToAnb → ResponseAggregated → Done
/// ```
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(tag = "status", content = "data")]
pub enum AktivierungState {
    /// No events yet.
    #[default]
    New,
    /// ACO received; response not yet sent.
    ActivationOrdered(AcoData),
    /// ACO cascaded to ANB (VNB role); awaiting response.
    DispatchedToAnb(AcoData),
    /// ANB/VNB confirmed full activation (ACR).
    Confirmed {
        /// Core activation data.
        data: AcoData,
        /// MRID of the sent ACR.
        acr_mrid: String,
    },
    /// ANB/VNB partially rejected (AAR).
    PartialRejection {
        /// Core activation data.
        data: AcoData,
        /// MRID of the sent AAR.
        aar_mrid: String,
    },
    /// VNB aggregated ANB responses and forwarded upstream.
    Done(AcoData),
    /// Activation deadline expired without a response.
    DeadlineExpired {
        /// Human-readable reason.
        reason: String,
    },
}

impl AktivierungState {
    /// Stable string label for the current variant.
    #[must_use]
    pub fn label(&self) -> &'static str {
        match self {
            Self::New => "New",
            Self::ActivationOrdered(_) => "ActivationOrdered",
            Self::DispatchedToAnb(_) => "DispatchedToAnb",
            Self::Confirmed { .. } => "Confirmed",
            Self::PartialRejection { .. } => "PartialRejection",
            Self::Done(_) => "Done",
            Self::DeadlineExpired { .. } => "DeadlineExpired",
        }
    }
}

// ── Commands ──────────────────────────────────────────────────────────────────

/// Commands for the Aktivierung workflow.
///
/// All domain values must be pre-extracted by the transport layer.
/// `Workflow::handle` is pure — no I/O.
#[derive(Clone)]
pub enum AktivierungCommand {
    /// Inbound ACO (`ActivationDocument` A96 Ordered) received and parsed.
    ReceiveAco {
        /// MRID of the ACO document.
        mrid: String,
        /// Aufforderungsfall vs Duldungsfall — from the resource's `Stammdaten`
        /// (`StatusDuldungsfall`/`AbrufartAufforderungsfall`), resolved by the
        /// transport layer before the command is issued.
        abwicklung: Abwicklung,
        /// Ordered MW.
        ordered_mw: f64,
        /// Resource object identifier.
        resource_id: String,
        /// Activation time period.
        period: String,
        /// Sender GLN.
        sender: String,
        /// Receiver GLN.
        receiver: String,
        /// UTC receipt timestamp.
        received_at: String,
    },
    /// Cascade ACO to ANB (VNB role only).
    CascadeToAnb {
        /// GLN of the target ANB.
        recipient_anb: String,
        /// MRID assigned to the outbound ACO.
        child_mrid: String,
    },
    /// Send ACR — full activation confirmed (ANB or VNB→ÜNB role).
    ///
    /// The caller is responsible for enqueuing the outbound XML via the outbox.
    SendAcr {
        /// MRID assigned to the outbound ACR.
        acr_mrid: String,
        /// MW actually activated.
        activated_mw: f64,
    },
    /// Send AAR — partial rejection (ANB or VNB→ÜNB role).
    ///
    /// The caller is responsible for enqueuing the outbound XML via the outbox.
    SendAar {
        /// MRID assigned to the outbound AAR.
        aar_mrid: String,
        /// MW available.
        available_mw: f64,
        /// Rejection reason code.
        reason_code: String,
    },
    /// Record ANB response received (VNB role).
    RecordAnbResponse {
        /// GLN of the responding ANB.
        anb_id: String,
        /// Confirmed or partially rejected.
        response_type: ResponseType,
        /// MRID of the ANB's ACR or AAR.
        response_mrid: String,
    },
    /// Forward aggregated response upstream to ÜNB (VNB role).
    ForwardResponse {
        /// MRID of the upstream ACR or AAR.
        upstream_mrid: String,
        /// Final aggregate response type.
        response_type: ResponseType,
    },
    /// Any other Redispatch EDIFACT market message received
    /// (MSCONS 13020/13021/13022/13023/13026, ORDERS 17209–17211,
    /// ORDRSP 19204/19301/19302). Audit-only — drives no transition, but
    /// gives every registered PID a handler instead of a silent route.
    ReceiveMarketMessage {
        /// The message's Prüfidentifikator.
        pid: u32,
        /// Sender MP-ID.
        sender: String,
        /// Receiver MP-ID.
        receiver: String,
        /// EDIFACT message reference.
        message_ref: String,
    },
    /// `AcknowledgementDocument` received for a document this process sent
    /// (correlation-routed via `ReceivingDocumentIdentification`). Audit
    /// record — the ACK deadline simply stops mattering once this lands.
    ReceiveAck {
        /// MRID of the `AcknowledgementDocument`.
        ack_mrid: String,
        /// The `ReceivingDocumentIdentification` it acknowledges.
        acknowledged_mrid: String,
        /// A01 accepted / A02 rejected.
        reason_code: String,
    },
    /// IFTSTA Vollzugsmeldung received (PID 21037 or 21038).
    ReceiveIftsta {
        /// IFTSTA PID (21037 = NB view, 21038 = BTR view).
        pid: u32,
        /// Sender GLN.
        sender: String,
        /// Receiver GLN.
        receiver: String,
        /// EDIFACT message reference.
        message_ref: String,
    },
    /// Deadline fired.
    TimeoutExpired {
        /// Unique ID of the expired deadline.
        deadline_id: DeadlineId,
        /// Label identifying the deadline type.
        label: Box<str>,
    },
}

impl CommandPayload for AktivierungCommand {}

// ── Workflow ──────────────────────────────────────────────────────────────────

/// Redispatch-Aktivierung workflow for Redispatch 2.0.
///
/// Handles ACO reception, response dispatch (ACR/AAR), VNB cascading,
/// and IFTSTA Vollzugsmeldung recording.
///
/// # Deadline requirements
///
/// The `makod` daemon must configure a **separate `DeadlineScheduler`** with a
/// ≤ 30 s tick interval for Redispatch workflows to honour the 5-minute ACO
/// response window ([`crate::fristen::`Betreiberfristen`::aktivierung_antwort`]).
/// The standard Werktage scheduler used
/// for GPKE/WiM is not sufficient.
///
/// Spawn via [`mako_engine::process::Process`]:
/// ```rust,ignore
/// let process = ctx.spawn::<AktivierungWorkflow>(
///     tenant_id,
///     WorkflowId::new(WORKFLOW_NAME, "FV2025-10-01"),
/// );
/// ```
pub struct AktivierungWorkflow;

impl Workflow for AktivierungWorkflow {
    type State = AktivierungState;
    type Event = AktivierungEvent;
    type Command = AktivierungCommand;

    fn on_deadline(deadline: &Deadline, state: &Self::State) -> Option<Self::Command> {
        match (deadline.label(), state) {
            // ACR/AAR response window: only meaningful in the
            // Aufforderungsfall — in the Duldungsfall the NB steers the
            // resource itself and awaits no counterparty response, so a
            // mistakenly scheduled window is ignored here.
            (
                ACTIVATION_RESPONSE_WINDOW_LABEL,
                AktivierungState::ActivationOrdered(data) | AktivierungState::DispatchedToAnb(data),
            ) if data.abwicklung.response_window_applies() => {
                Some(AktivierungCommand::TimeoutExpired {
                    deadline_id: deadline.deadline_id(),
                    label: deadline.label().into(),
                })
            }
            // 3-minute AcknowledgementDocument window: our sent
            // ACR/AAR was never technically acknowledged — escalate.
            (
                ACK_WINDOW_LABEL,
                AktivierungState::Confirmed { .. } | AktivierungState::PartialRejection { .. },
            ) => Some(AktivierungCommand::TimeoutExpired {
                deadline_id: deadline.deadline_id(),
                label: deadline.label().into(),
            }),
            _ => None,
        }
    }

    fn apply(state: Self::State, event: &Self::Event) -> Self::State {
        match event {
            AktivierungEvent::AcoReceived {
                mrid,
                abwicklung,
                ordered_mw,
                resource_id,
                period,
                sender,
                receiver,
                received_at,
            } => AktivierungState::ActivationOrdered(AcoData {
                mrid: mrid.clone(),
                abwicklung: *abwicklung,
                ordered_mw: *ordered_mw,
                resource_id: resource_id.clone(),
                period: period.clone(),
                sender: sender.clone(),
                receiver: receiver.clone(),
                received_at: received_at.clone(),
            }),

            AktivierungEvent::AcoCascaded { .. } => match state {
                AktivierungState::ActivationOrdered(data) => {
                    AktivierungState::DispatchedToAnb(data)
                }
                other => other,
            },

            AktivierungEvent::AcrSent { acr_mrid, .. } => match state {
                AktivierungState::ActivationOrdered(data)
                | AktivierungState::DispatchedToAnb(data) => AktivierungState::Confirmed {
                    data,
                    acr_mrid: acr_mrid.clone(),
                },
                other => other,
            },

            AktivierungEvent::AarSent { aar_mrid, .. } => match state {
                AktivierungState::ActivationOrdered(data)
                | AktivierungState::DispatchedToAnb(data) => AktivierungState::PartialRejection {
                    data,
                    aar_mrid: aar_mrid.clone(),
                },
                other => other,
            },

            AktivierungEvent::AnbResponseReceived { .. } => {
                // Does not change the VNB's own state variant — tracked via projection.
                state
            }

            AktivierungEvent::ResponseForwarded { .. } => match state {
                AktivierungState::DispatchedToAnb(data) => AktivierungState::Done(data),
                other => other,
            },

            // Audit-only events — do not drive state transitions.
            AktivierungEvent::IftstaReceived { .. }
            | AktivierungEvent::AckReceived { .. }
            | AktivierungEvent::MarketMessageReceived { .. } => state,

            AktivierungEvent::DeadlineExpired { label, .. } => {
                let is_ack_window = &**label == ACK_WINDOW_LABEL;
                match state {
                    // ACK window: an unacknowledged ACR/AAR is a terminal
                    // error even from Confirmed/PartialRejection.
                    AktivierungState::Confirmed { .. }
                    | AktivierungState::PartialRejection { .. }
                        if is_ack_window =>
                    {
                        AktivierungState::DeadlineExpired {
                            reason: format!(
                                "AcknowledgementDocument not received within the 3-minute window ({label})"
                            ),
                        }
                    }
                    AktivierungState::Confirmed { .. }
                    | AktivierungState::PartialRejection { .. }
                    | AktivierungState::Done(_)
                    | AktivierungState::DeadlineExpired { .. } => state,
                    _ => AktivierungState::DeadlineExpired {
                        reason: format!("deadline expired: {label}"),
                    },
                }
            }
        }
    }

    // Workflow state machines legitimately require large match expressions.
    // Each arm is a single documented state-transition rule.
    #[allow(clippy::too_many_lines)]
    fn handle(
        state: &Self::State,
        command: Self::Command,
    ) -> Result<WorkflowOutput<Self::Event>, WorkflowError> {
        match command {
            AktivierungCommand::ReceiveAco {
                mrid,
                abwicklung,
                ordered_mw,
                resource_id,
                period,
                sender,
                receiver,
                received_at,
            } => {
                if !matches!(state, AktivierungState::New) {
                    return Ok(vec![].into());
                }
                Ok(vec![AktivierungEvent::AcoReceived {
                    mrid,
                    abwicklung,
                    ordered_mw,
                    resource_id,
                    period,
                    sender,
                    receiver,
                    received_at,
                }]
                .into())
            }

            AktivierungCommand::CascadeToAnb {
                recipient_anb,
                child_mrid,
            } => match state {
                AktivierungState::ActivationOrdered(_) => Ok(vec![AktivierungEvent::AcoCascaded {
                    recipient_anb,
                    child_mrid,
                }]
                .into()),
                AktivierungState::DispatchedToAnb(_) => Ok(vec![].into()),
                other => Err(WorkflowError::rejected(format!(
                    "CascadeToAnb not valid in state {}",
                    other.label()
                ))),
            },

            AktivierungCommand::SendAcr {
                acr_mrid,
                activated_mw,
            } => match state {
                AktivierungState::ActivationOrdered(_) | AktivierungState::DispatchedToAnb(_) => {
                    Ok(vec![AktivierungEvent::AcrSent {
                        acr_mrid,
                        activated_mw,
                    }]
                    .into())
                }
                AktivierungState::Confirmed { .. } => Ok(vec![].into()),
                other => Err(WorkflowError::rejected(format!(
                    "SendAcr not valid in state {}",
                    other.label()
                ))),
            },

            AktivierungCommand::SendAar {
                aar_mrid,
                available_mw,
                reason_code,
            } => match state {
                AktivierungState::ActivationOrdered(_) | AktivierungState::DispatchedToAnb(_) => {
                    Ok(vec![AktivierungEvent::AarSent {
                        aar_mrid,
                        available_mw,
                        reason_code,
                    }]
                    .into())
                }
                AktivierungState::PartialRejection { .. } => Ok(vec![].into()),
                other => Err(WorkflowError::rejected(format!(
                    "SendAar not valid in state {}",
                    other.label()
                ))),
            },

            AktivierungCommand::RecordAnbResponse {
                anb_id,
                response_type,
                response_mrid,
            } => match state {
                AktivierungState::DispatchedToAnb(_) => {
                    Ok(vec![AktivierungEvent::AnbResponseReceived {
                        anb_id,
                        response_type,
                        response_mrid,
                    }]
                    .into())
                }
                other => Err(WorkflowError::rejected(format!(
                    "RecordAnbResponse not valid in state {}",
                    other.label()
                ))),
            },

            AktivierungCommand::ForwardResponse {
                upstream_mrid,
                response_type,
            } => match state {
                AktivierungState::DispatchedToAnb(_) => {
                    Ok(vec![AktivierungEvent::ResponseForwarded {
                        upstream_mrid,
                        response_type,
                    }]
                    .into())
                }
                AktivierungState::Done(_) => Ok(vec![].into()),
                other => Err(WorkflowError::rejected(format!(
                    "ForwardResponse not valid in state {}",
                    other.label()
                ))),
            },

            AktivierungCommand::ReceiveAck {
                ack_mrid,
                acknowledged_mrid,
                reason_code,
            } => Ok(vec![AktivierungEvent::AckReceived {
                ack_mrid,
                acknowledged_mrid,
                reason_code,
            }]
            .into()),

            // MSCONS/ORDERS/ORDRSP: audit record in any state — every
            // registered PID now has a handler.
            AktivierungCommand::ReceiveMarketMessage {
                pid,
                sender,
                receiver,
                message_ref,
            } => Ok(vec![AktivierungEvent::MarketMessageReceived {
                pid,
                sender,
                receiver,
                message_ref,
            }]
            .into()),

            // IFTSTA is accepted in any non-terminal state — audit record only.
            AktivierungCommand::ReceiveIftsta {
                pid,
                sender,
                receiver,
                message_ref,
            } => Ok(vec![AktivierungEvent::IftstaReceived {
                pid,
                sender,
                receiver,
                message_ref,
            }]
            .into()),

            AktivierungCommand::TimeoutExpired { deadline_id, label } => {
                match (state, &*label) {
                    // ACK escalation is valid from a sent-response state.
                    (
                        AktivierungState::Confirmed { .. }
                        | AktivierungState::PartialRejection { .. },
                        ACK_WINDOW_LABEL,
                    ) => Ok(vec![AktivierungEvent::DeadlineExpired { deadline_id, label }].into()),
                    (
                        AktivierungState::Confirmed { .. }
                        | AktivierungState::PartialRejection { .. }
                        | AktivierungState::Done(_)
                        | AktivierungState::DeadlineExpired { .. },
                        _,
                    ) => Ok(vec![].into()),
                    _ => Ok(vec![AktivierungEvent::DeadlineExpired { deadline_id, label }].into()),
                }
            }
        }
    }
}

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

    fn aco_data() -> AcoData {
        AcoData {
            abwicklung: Abwicklung::Aufforderungsfall {
                abrufart: Abrufart::Sollwert,
            },
            mrid: "aco-001".into(),
            ordered_mw: 50.0,
            resource_id: "4012345678901".into(),
            period: "2025-10-15T10:00:00Z/2025-10-15T11:00:00Z".into(),
            sender: "4012345000001".into(),
            receiver: "4012345000002".into(),
            received_at: "2025-10-15T09:58:00Z".into(),
        }
    }

    fn test_deadline(label: &str) -> Deadline {
        use mako_engine::ids::{ProcessId, StreamId, TenantId};
        use mako_engine::version::WorkflowId;
        Deadline::new(
            StreamId::new("process/test"),
            ProcessId::new(),
            TenantId::new(),
            WorkflowId::new(WORKFLOW_NAME, "FV2025-10-01"),
            label,
            time::OffsetDateTime::now_utc(),
        )
    }

    fn receive_aco_cmd() -> AktivierungCommand {
        AktivierungCommand::ReceiveAco {
            abwicklung: Abwicklung::Aufforderungsfall {
                abrufart: Abrufart::Sollwert,
            },
            mrid: "aco-001".into(),
            ordered_mw: 50.0,
            resource_id: "4012345678901".into(),
            period: "2025-10-15T10:00:00Z/2025-10-15T11:00:00Z".into(),
            sender: "4012345000001".into(),
            receiver: "4012345000002".into(),
            received_at: "2025-10-15T09:58:00Z".into(),
        }
    }

    #[test]
    fn receive_aco_transitions_new_to_activation_ordered() {
        let state = AktivierungState::New;
        let output = AktivierungWorkflow::handle(&state, receive_aco_cmd()).unwrap();
        assert_eq!(output.events.len(), 1);
        let new_state = AktivierungWorkflow::apply(state, &output.events[0]);
        assert!(matches!(new_state, AktivierungState::ActivationOrdered(_)));
    }

    #[test]
    fn send_acr_from_activation_ordered_produces_confirmed() {
        let state = AktivierungState::ActivationOrdered(aco_data());
        let output = AktivierungWorkflow::handle(
            &state,
            AktivierungCommand::SendAcr {
                acr_mrid: "acr-001".into(),
                activated_mw: 50.0,
            },
        )
        .unwrap();
        let new_state = AktivierungWorkflow::apply(state, &output.events[0]);
        assert!(matches!(new_state, AktivierungState::Confirmed { .. }));
    }

    #[test]
    fn send_aar_from_activation_ordered_produces_partial_rejection() {
        let state = AktivierungState::ActivationOrdered(aco_data());
        let output = AktivierungWorkflow::handle(
            &state,
            AktivierungCommand::SendAar {
                aar_mrid: "aar-001".into(),
                available_mw: 30.0,
                reason_code: "A96".into(),
            },
        )
        .unwrap();
        let new_state = AktivierungWorkflow::apply(state, &output.events[0]);
        assert!(matches!(
            new_state,
            AktivierungState::PartialRejection { .. }
        ));
    }

    #[test]
    fn timeout_in_confirmed_state_is_noop() {
        let state = AktivierungState::Confirmed {
            data: aco_data(),
            acr_mrid: "acr-001".into(),
        };
        let output = AktivierungWorkflow::handle(
            &state,
            AktivierungCommand::TimeoutExpired {
                deadline_id: DeadlineId::new(),
                label: ACTIVATION_RESPONSE_WINDOW_LABEL.into(),
            },
        )
        .unwrap();
        assert!(output.events.is_empty());
    }

    #[test]
    fn ack_received_is_audit_only_and_accepted_in_any_state() {
        for state in [
            AktivierungState::New,
            AktivierungState::ActivationOrdered(aco_data()),
            AktivierungState::Confirmed {
                data: aco_data(),
                acr_mrid: "acr-001".into(),
            },
        ] {
            let output = AktivierungWorkflow::handle(
                &state,
                AktivierungCommand::ReceiveAck {
                    ack_mrid: "ack-001".into(),
                    acknowledged_mrid: "aco-001".into(),
                    reason_code: String::new(),
                },
            )
            .unwrap();
            assert!(matches!(
                output.events.as_slice(),
                [AktivierungEvent::AckReceived { .. }]
            ));
            // Audit-only: the ACK never changes the process state.
            let before = format!("{state:?}");
            let after = AktivierungWorkflow::apply(state, &output.events[0]);
            assert_eq!(before, format!("{after:?}"));
        }
    }

    #[test]
    fn iftsta_accepted_in_any_state() {
        for state in [
            AktivierungState::New,
            AktivierungState::ActivationOrdered(aco_data()),
            AktivierungState::Confirmed {
                data: aco_data(),
                acr_mrid: "x".into(),
            },
        ] {
            let output = AktivierungWorkflow::handle(
                &state,
                AktivierungCommand::ReceiveIftsta {
                    pid: 21_037,
                    sender: "s".into(),
                    receiver: "r".into(),
                    message_ref: "ref-1".into(),
                },
            )
            .unwrap();
            assert!(matches!(
                output.events.as_slice(),
                [AktivierungEvent::IftstaReceived { pid: 21037, .. }]
            ));
        }
    }

    #[test]
    fn duldungsfall_ignores_the_response_window_deadline() {
        // NB steers the resource itself — no counterparty response is
        // awaited, so the 5-minute window must not expire the process.
        let mut data = aco_data();
        data.abwicklung = Abwicklung::Duldungsfall;
        let state = AktivierungState::ActivationOrdered(data);
        let deadline = test_deadline(ACTIVATION_RESPONSE_WINDOW_LABEL);
        assert!(
            AktivierungWorkflow::on_deadline(&deadline, &state).is_none(),
            "Duldungsfall must not time out waiting for a response"
        );
        // Aufforderungsfall keeps the hard 5-minute constraint.
        let state2 = AktivierungState::ActivationOrdered(aco_data());
        assert!(AktivierungWorkflow::on_deadline(&deadline, &state2).is_some());
    }

    #[test]
    fn unacknowledged_response_escalates_after_the_ack_window() {
        let state = AktivierungState::Confirmed {
            data: aco_data(),
            acr_mrid: "ACR-1".into(),
        };
        let deadline = test_deadline(ACK_WINDOW_LABEL);
        let cmd = AktivierungWorkflow::on_deadline(&deadline, &state)
            .expect("ACK window fires from Confirmed");
        let out = AktivierungWorkflow::handle(&state, cmd).expect("timeout handled");
        let next = out.events.iter().fold(state, AktivierungWorkflow::apply);
        assert!(
            matches!(&next, AktivierungState::DeadlineExpired { reason } if reason.contains("Acknowledgement")),
            "got {next:?}"
        );
    }

    #[test]
    fn market_messages_are_audited_in_any_state() {
        let state = AktivierungState::Confirmed {
            data: aco_data(),
            acr_mrid: "ACR-1".into(),
        };
        let out = AktivierungWorkflow::handle(
            &state,
            AktivierungCommand::ReceiveMarketMessage {
                pid: 13_020,
                sender: "9900000000001".into(),
                receiver: "9900000000002".into(),
                message_ref: "MSG-1".into(),
            },
        )
        .expect("audit accepted");
        assert_eq!(out.events.len(), 1);
        // No state transition.
        let next = out.events.iter().fold(state, AktivierungWorkflow::apply);
        assert!(matches!(next, AktivierungState::Confirmed { .. }));
    }
}