obzenflow_runtime 0.2.4

// SPDX-License-Identifier: MIT OR Apache-2.0
// SPDX-FileCopyrightText: 2025-2026 ObzenFlow Contributors
// https://obzenflow.dev

use super::*;
use obzenflow_core::event::ChainPayload;

/// Structurally metadata-only view used before effect-boundary admission.
///
/// It deliberately exposes neither callable ports nor registry or resolver
/// authority. Metadata is co-resolved with its callable port under the same
/// run-scoped verdict.
#[derive(Clone, Default)]
pub struct EffectPortMetadataContext {
    pub(super) metadata: super::ports::EffectPortMetadataView,
}

impl EffectPortMetadataContext {
    /// Return the immutable snapshot for one declared metadata-bearing slot.
    pub fn metadata<P, M>(&self, slot: EffectPortSlot<P, M>) -> Result<Arc<M>, EffectError>
    where
        P: ?Sized + Send + Sync + 'static,
        M: Send + Sync + 'static,
    {
        self.metadata
            .get(slot)
            .ok_or_else(|| EffectError::target_invariant_violation(slot))
    }
}

impl std::fmt::Debug for EffectPortMetadataContext {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("EffectPortMetadataContext")
            .field("metadata", &"<not disclosed>")
            .finish()
    }
}

#[derive(Clone)]
pub struct EffectContext {
    pub(super) is_replaying: bool,
    pub(super) flow_id: FlowId,
    pub(super) stage_key: String,
    pub(super) input_seq: StageInputPosition,
    pub(super) ports: super::ports::EffectPortView,
}

impl EffectContext {
    pub fn is_replaying(&self) -> bool {
        self.is_replaying
    }

    pub fn flow_id(&self) -> FlowId {
        self.flow_id
    }

    pub fn stage_key(&self) -> &str {
        &self.stage_key
    }

    pub fn input_seq(&self) -> StageInputPosition {
        self.input_seq
    }

    pub fn now(&self) -> u64 {
        self.input_seq.0
    }

    pub fn deterministic_id(
        &self,
        label: &str,
        ordinal: impl Into<EffectOutputOrdinal>,
    ) -> EventId {
        deterministic_event_id(
            self.flow_id.to_string(),
            format!("{}:{label}", self.stage_key),
            self.input_seq,
            ordinal,
        )
    }

    pub fn rng(&self, label: &str) -> fastrand::Rng {
        let material = format!(
            "{}:{}:{}:{label}",
            self.flow_id, self.stage_key, self.input_seq.0
        );
        let hash = digest(&SHA256, material.as_bytes());
        let mut seed = [0u8; 8];
        seed.copy_from_slice(&hash.as_ref()[..8]);
        fastrand::Rng::with_seed(u64::from_be_bytes(seed))
    }

    pub fn port<T, M>(&self, slot: EffectPortSlot<T, M>) -> Result<Arc<T>, EffectError>
    where
        T: ?Sized + Send + Sync + 'static,
        M: Send + Sync + 'static,
    {
        self.ports
            .get(slot)
            .ok_or_else(|| EffectError::target_invariant_violation(slot))
    }

    pub fn sleep(&self, duration: Duration) -> impl std::future::Future<Output = ()> + Send {
        tokio::time::sleep(duration)
    }
}

pub struct EffectInvocationContext {
    pub flow_id: FlowId,
    pub stage_id: StageId,
    pub stage_key: String,
    pub writer_id: WriterId,
    pub input_seq: StageInputPosition,
    pub stage_logic_version: String,
    pub data_journal: Arc<dyn Journal<ChainEvent>>,
    pub flow_context: Option<FlowContext>,
    pub observers: Option<crate::stages::observer::StageObserverBundle>,
    pub system_journal: Option<Arc<dyn Journal<SystemEvent>>>,
    pub instrumentation: Option<Arc<StageInstrumentation>>,
    pub heartbeat_state: Option<Arc<HeartbeatState>>,
    pub parent: JournalRecord<ChainPayload>,
    pub effect_history: Option<Arc<EffectHistory>>,
    /// Runtime execution strategy (FLOWIP-120r): one authority for the
    /// replay-versus-live decision at the effect boundary.
    pub runtime_execution: crate::execution::RuntimeExecution,
    pub effect_ports: EffectPortRegistry,
    pub effect_declarations: Vec<EffectDeclaration>,
    pub output_contract: StageOutputContract,
    pub backpressure_writer: BackpressureWriter,
    pub emit_enabled: bool,
    pub effect_boundary: Option<Arc<dyn EffectBoundary>>,
    /// FLOWIP-010 §7: build-resolved lineage policy from stage resources,
    /// consumed as data when effect facts derive from the parent event.
    pub lineage: obzenflow_core::config::LineagePolicy,
}

impl EffectInvocationContext {
    pub fn effect_declaration(
        &self,
        effect_type: &'static str,
    ) -> Result<EffectDeclaration, EffectError> {
        let reported_effect_type = if super::binding::validate_effect_type(effect_type).is_ok() {
            effect_type
        } else {
            "invalid_effect_type"
        };
        self.effect_declarations
            .iter()
            .find(|declaration| declaration.effect_type() == effect_type)
            .cloned()
            .ok_or_else(|| EffectError::UndeclaredEffect {
                stage_key: self.stage_key.clone(),
                effect_type: reported_effect_type.to_string(),
            })
    }
}

// FLOWIP-120r: the runtime execution strategy (`crate::execution`) is now the
// single replay-versus-live authority. `EffectRuntimeMode` survives only as a
// test parameterization aid (the `RuntimeExecution::from_effect_runtime_mode`
// bridge maps it to a strategy); the per-dispatch scope helper, the
// archive-to-mode derivation, and the scope `From` impl are gone, replaced by
// the strategy.
#[cfg(test)]
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub enum EffectRuntimeMode {
    #[default]
    Live,
    ReplayStrict,
    ResumeIncomplete,
}

// FLOWIP-120r: the former `resume_scope_remains_reconstruction_until_phase_predicate_lands`
// test moved to `crate::execution` (the strategy policy matrix), which now owns
// the replay-incomplete -> ResumeHandler mapping.