meerkat-mob 0.7.1

//! Error types for mob operations.

use crate::ids::{AgentIdentity, AgentRuntimeId, FenceToken, FlowId, LoopId, ProfileName, WorkRef};
use crate::runtime::MobState;
use crate::store::FrameAtomicOperation;
use crate::validate::Diagnostic;
use crate::{MobId, RunId, StepId};
use meerkat_contracts::wire::supervisor_bridge::{BridgeRejectionCause, BridgeRejectionReply};

/// Runtime capability required from a seated mob member.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MobMemberCapability {
    /// Interaction-scoped injection used for autonomous console/RPC/flow turns.
    InteractionEventInjector,
}

/// Mob-owned classification of why a mob operation failed.
///
/// This is the typed owner of "what class of failure does this `MobError`
/// represent" — the knowledge of which variants are missing-target vs busy vs
/// transport vs internal lives next to the variants themselves, not in
/// downstream classifiers that re-`match` the enum.
///
/// Consumers (e.g. the schedule delivery host) map this onto their own
/// domain-failure vocabulary (`DeliveryFailureReason`); the mapping lives at
/// the consumer because the schedule-failure type is owned by a crate
/// `meerkat-mob` does not (and must not) depend on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MobFailureClass {
    /// The addressed mob/profile/member/flow/run/work does not exist.
    TargetMissing,
    /// The target exists but cannot accept the operation right now
    /// (e.g. a member id collision).
    TargetBusy,
    /// A transport/persistence/session/timeout fault prevented delivery.
    Transport,
    /// A runtime accepted the work but parked at an external boundary
    /// (callback pending) rather than completing.
    RuntimeRejected,
    /// An internal/unexpected-state fault.
    Internal,
    /// The mob authority rejected the operation on its own terms.
    MobRejected,
}

impl std::fmt::Display for MobMemberCapability {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::InteractionEventInjector => f.write_str("interaction_event_injector"),
        }
    }
}

/// Errors returned by mob operations.
#[derive(Debug, thiserror::Error)]
pub enum MobError {
    /// The requested mob does not exist in the runtime/store.
    #[error("mob not found: {0}")]
    MobNotFound(MobId),

    /// The requested profile does not exist in the mob definition.
    #[error("profile not found: {0}")]
    ProfileNotFound(ProfileName),

    /// The requested mob member does not exist in the roster.
    #[error("mob member not found: {0}")]
    MemberNotFound(AgentIdentity),

    /// A mob member with the given ID already exists.
    #[error("mob member already exists: {0}")]
    MemberAlreadyExists(AgentIdentity),

    /// The mob member's profile does not allow external turns.
    #[error("mob member is not externally addressable: {0}")]
    NotExternallyAddressable(AgentIdentity),

    /// The requested lifecycle state transition is invalid.
    #[error("invalid state transition: {from} -> {to}")]
    InvalidTransition { from: MobState, to: MobState },

    /// A wiring operation failed.
    #[error("wiring error: {0}")]
    WiringError(String),

    /// Supervisor rotation reached one or more remote members but did not
    /// complete, so local supervisor authority stayed at the pre-rotation
    /// epoch.
    #[error(
        "supervisor rotation incomplete: failed after {rotated_peer_count} remote peer(s) accepted attempted epoch {attempted_epoch}; local authority remains at epoch {previous_epoch}; rollback_succeeded={rollback_succeeded}; pending_authority_recorded={pending_authority_recorded}; failure: {reason}"
    )]
    SupervisorRotationIncomplete {
        previous_epoch: u64,
        attempted_epoch: u64,
        attempted_public_peer_id: String,
        rotated_peer_count: usize,
        rollback_succeeded: bool,
        pending_authority_recorded: bool,
        rollback_error: Option<String>,
        reason: String,
    },

    /// A supervisor bridge command was rejected by the remote member.
    #[error("bridge command rejected ({cause:?}): {reason}")]
    BridgeCommandRejected {
        cause: BridgeRejectionCause,
        reason: String,
    },

    /// The member failed to restore durable session state and is broken until repaired.
    #[error(
        "member {member_id} failed to restore {}: {reason}",
        format_member_restore_target(.session_id.as_ref())
    )]
    MemberRestoreFailed {
        member_id: AgentIdentity,
        session_id: Option<meerkat_core::types::SessionId>,
        reason: String,
    },

    /// Waiting for kickoff completion timed out.
    #[error("kickoff wait timed out")]
    KickoffWaitTimedOut {
        pending_member_ids: Vec<AgentIdentity>,
    },

    /// Waiting for startup readiness timed out.
    #[error("member ready wait timed out")]
    ReadyWaitTimedOut {
        pending_member_ids: Vec<AgentIdentity>,
    },

    /// The mob definition failed validation.
    #[error("definition error: {}", format_diagnostics(.0))]
    DefinitionError(Vec<Diagnostic>),

    /// Referenced flow does not exist.
    #[error("flow not found: {0}")]
    FlowNotFound(FlowId),

    /// Run failed with a reason.
    #[error("flow failed for run {run_id}: {reason}")]
    FlowFailed { run_id: RunId, reason: String },

    /// Referenced run does not exist.
    #[error("run not found: {0}")]
    RunNotFound(RunId),

    /// Run was canceled.
    #[error("run canceled: {0}")]
    RunCanceled(RunId),

    /// Flow turn timed out while awaiting terminal transport outcome.
    #[error("flow turn timed out")]
    FlowTurnTimedOut,

    /// A frame-aware flow exceeded its configured nesting depth.
    #[error(
        "loop '{loop_id}' would exceed max_frame_depth={max_frame_depth} (current depth={current_depth})"
    )]
    FrameDepthLimitExceeded {
        loop_id: LoopId,
        max_frame_depth: u32,
        current_depth: u32,
    },

    /// The selected mob run store cannot provide frame-aware atomic persistence.
    #[error("mob run store cannot atomically persist frame operation '{operation}'")]
    FrameAtomicPersistenceUnavailable { operation: FrameAtomicOperation },

    /// Spec revision compare-and-swap failed.
    #[error("spec revision conflict for mob {mob_id}: expected {expected:?}, actual {actual}")]
    SpecRevisionConflict {
        mob_id: MobId,
        expected: Option<u64>,
        actual: u64,
    },

    /// Schema validation failed for a step output.
    #[error("schema validation failed for step {step_id}: {message}")]
    SchemaValidation { step_id: StepId, message: String },

    /// Not enough targets to satisfy dispatch/collection policy.
    #[error("insufficient targets for step {step_id}: required {required}, available {available}")]
    InsufficientTargets {
        step_id: StepId,
        required: u8,
        available: usize,
    },

    /// Topology policy denied a dispatch edge.
    #[error("topology violation: {from_role} -> {to_role}")]
    TopologyViolation {
        from_role: ProfileName,
        to_role: ProfileName,
    },

    /// A bridge accepted the delivery command but rejected the member input.
    #[error("bridge delivery rejected ({cause}): {reason}")]
    BridgeDeliveryRejected {
        cause: meerkat_contracts::wire::supervisor_bridge::BridgeDeliveryRejectionCause,
        reason: String,
    },

    /// Supervisor escalation happened.
    #[error("supervisor escalation: {0}")]
    SupervisorEscalation(String),

    /// Operation is not supported for the member's runtime mode.
    #[error("unsupported for runtime mode {mode}: {reason}")]
    UnsupportedForMode {
        mode: crate::MobRuntimeMode,
        reason: String,
    },

    /// A member is missing a required runtime capability for the requested operation.
    #[error("mob member {member_id} missing required capability {capability}: {context}")]
    MissingMemberCapability {
        member_id: AgentIdentity,
        capability: MobMemberCapability,
        context: &'static str,
    },

    /// Operation blocked by reset barrier.
    #[error("reset barrier active")]
    ResetBarrier,

    /// A storage operation failed.
    #[error("storage error: {0}")]
    StorageError(#[source] Box<dyn std::error::Error + Send + Sync>),

    /// A session service operation failed.
    #[error("session error: {0}")]
    SessionError(#[from] meerkat_core::service::SessionError),

    /// A comms operation failed.
    #[error("comms error: {0}")]
    CommsError(#[from] meerkat_core::comms::SendError),

    /// A runtime-backed member turn reached an external callback boundary.
    #[error("callback pending for session {session_id} on tool '{tool_name}'")]
    CallbackPending {
        session_id: meerkat_core::types::SessionId,
        tool_name: String,
        args: serde_json::Value,
    },

    /// The fence token does not match the member's current incarnation.
    #[error("stale fence token for {runtime_id}: expected {expected}, got {actual}")]
    StaleFenceToken {
        runtime_id: AgentRuntimeId,
        expected: FenceToken,
        actual: FenceToken,
    },

    /// A caller supplied an event replay cursor beyond the store frontier.
    #[error("stale mob event cursor: requested {after_cursor}, latest {latest_cursor}")]
    StaleEventCursor {
        after_cursor: u64,
        latest_cursor: u64,
    },

    /// The referenced work unit does not exist.
    #[error("work not found: {0}")]
    WorkNotFound(WorkRef),

    /// Per-unit work cancellation is not realized: no work-tracking ledger
    /// backs `cancel_work`, so there is no authority that can locate and
    /// cancel an individual submitted unit. Returned instead of a phantom
    /// `WorkNotFound` so the advertised `mob/cancel_work` surface fails
    /// closed with an honest "unsupported" signal rather than claiming a
    /// search-and-miss that never happened. Callers that need to cancel
    /// in-flight work should use member-scoped `cancel_all_work`.
    #[error("per-unit work cancellation is unsupported for {0}: no work-tracking ledger is wired")]
    WorkCancellationUnsupported(WorkRef),

    /// The mob actor command channel closed before accepting a command.
    #[error("mob actor command channel closed")]
    ActorCommandChannelClosed,

    /// The mob actor accepted a command but dropped the reply channel.
    #[error("mob actor reply channel closed")]
    ActorReplyChannelClosed,

    /// A bridge session could not be located in any live mob authority roster
    /// (and is not owned by service-reported or persisted authority either),
    /// so live-handle retirement cannot resolve a member to retire.
    ///
    /// This is a typed recovery-class observation: callers that already hold
    /// independent evidence the bridge session is mob-owned (e.g. bridge-session
    /// scoped mobs) may proceed to scoped cleanup instead of failing.
    #[error("bridge session not found in any live mob authority: {bridge_session_id}")]
    BridgeSessionNotInLiveAuthority { bridge_session_id: String },

    /// A mob-member comms (peer) routing name could not be rendered because a
    /// component failed the [`meerkat_core::MemberCommsName`] slug rule.
    ///
    /// The routing-name shape has exactly one fail-closed owner; rendering
    /// surfaces the typed component fault here instead of reconstructing a raw
    /// `mob_id/role/member` join that the owner already rejected.
    #[error("invalid mob member comms name: {0}")]
    MemberCommsName(#[from] meerkat_core::MemberCommsNameError),

    /// A flow/loop condition could not be evaluated to a definite boolean
    /// because it referenced an absent/invalid context path or compared
    /// non-comparable operands.
    ///
    /// Surfaced as a typed fault (`location` names the owning step or loop)
    /// rather than silently evaluating the condition to `false` and skipping
    /// the step / mis-deciding the loop-until.
    #[error("condition evaluation failed for {location}: {reason}")]
    ConditionEval { location: String, reason: String },

    /// An internal error (unexpected state, logic errors).
    #[error("internal error: {0}")]
    Internal(String),
}

fn format_diagnostics(diagnostics: &[Diagnostic]) -> String {
    diagnostics
        .iter()
        .map(|d| format!("{}: {}", d.code, d.message))
        .collect::<Vec<_>>()
        .join("; ")
}

fn format_member_restore_target(session_id: Option<&meerkat_core::types::SessionId>) -> String {
    match session_id {
        Some(session_id) => format!("session {session_id}"),
        None => "runtime bridge state".to_string(),
    }
}

impl From<Box<dyn std::error::Error + Send + Sync>> for MobError {
    fn from(error: Box<dyn std::error::Error + Send + Sync>) -> Self {
        Self::StorageError(error)
    }
}

impl From<crate::store::MobStoreError> for MobError {
    fn from(error: crate::store::MobStoreError) -> Self {
        match error {
            crate::store::MobStoreError::SpecRevisionConflict {
                mob_id,
                expected,
                actual,
            } => Self::SpecRevisionConflict {
                mob_id,
                expected,
                actual,
            },
            crate::store::MobStoreError::FrameAtomicPersistenceUnavailable { operation } => {
                Self::FrameAtomicPersistenceUnavailable { operation }
            }
            other => Self::StorageError(Box::new(other)),
        }
    }
}

impl From<BridgeRejectionReply> for MobError {
    fn from(rejection: BridgeRejectionReply) -> Self {
        let cause = rejection.typed_cause();
        let reason = rejection.reason().to_string();
        match cause {
            Some(cause) => Self::BridgeCommandRejected { cause, reason },
            None => Self::WiringError(reason),
        }
    }
}

impl MobError {
    pub fn bridge_rejection_cause(&self) -> Option<BridgeRejectionCause> {
        match self {
            Self::BridgeCommandRejected { cause, .. } => Some(*cause),
            _ => None,
        }
    }

    /// Whether this error means the addressed target (mob, profile, member,
    /// flow, run, or work unit) does not exist.
    ///
    /// Owned here so target-existence probing does not re-`match` the
    /// `MobError` variant list in downstream classifiers.
    pub fn is_missing_target(&self) -> bool {
        matches!(self.failure_class(), MobFailureClass::TargetMissing)
    }

    /// Classify this error into the mob-owned [`MobFailureClass`].
    ///
    /// This is the single source of truth for which `MobError` variants fall
    /// into which failure class; consumers map [`MobFailureClass`] onto their
    /// own domain vocabularies rather than re-matching the variant list.
    pub fn failure_class(&self) -> MobFailureClass {
        match self {
            Self::MobNotFound(_)
            | Self::ProfileNotFound(_)
            | Self::MemberNotFound(_)
            | Self::FlowNotFound(_)
            | Self::RunNotFound(_)
            | Self::WorkNotFound(_) => MobFailureClass::TargetMissing,
            Self::MemberAlreadyExists(_) => MobFailureClass::TargetBusy,
            Self::StorageError(_)
            | Self::SessionError(_)
            | Self::CommsError(_)
            | Self::MemberRestoreFailed { .. }
            | Self::KickoffWaitTimedOut { .. }
            | Self::ReadyWaitTimedOut { .. }
            | Self::FlowTurnTimedOut => MobFailureClass::Transport,
            Self::Internal(_) => MobFailureClass::Internal,
            Self::CallbackPending { .. } => MobFailureClass::RuntimeRejected,
            _ => MobFailureClass::MobRejected,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::validate::{Diagnostic, DiagnosticCode, DiagnosticSeverity};

    #[test]
    fn test_profile_not_found_display() {
        let err = MobError::ProfileNotFound(ProfileName::from("missing"));
        assert!(format!("{err}").contains("missing"));
    }

    /// DELETE_ME A2 + B8 regression: the `Meerkat*` variant prefix and
    /// the "meerkat" literal in error messages were renamed to
    /// identity-first terminology ("mob member"). This test pins both
    /// the display-string and the variant-construction shape so the
    /// 0.6 identity-first cascade cannot regress into legacy wording.
    #[test]
    fn member_not_found_and_already_exists_use_identity_first_display() {
        let not_found = MobError::MemberNotFound(AgentIdentity::from("singer"));
        let already = MobError::MemberAlreadyExists(AgentIdentity::from("singer"));
        let not_addressable = MobError::NotExternallyAddressable(AgentIdentity::from("singer"));

        let msg_nf = format!("{not_found}");
        let msg_ae = format!("{already}");
        let msg_na = format!("{not_addressable}");

        assert_eq!(msg_nf, "mob member not found: singer");
        assert_eq!(msg_ae, "mob member already exists: singer");
        assert_eq!(msg_na, "mob member is not externally addressable: singer");

        // No legacy "meerkat" literal should appear in any of the
        // identity-first error displays.
        for msg in [&msg_nf, &msg_ae, &msg_na] {
            assert!(
                !msg.to_lowercase().contains("meerkat"),
                "identity-first mob errors must not carry legacy 'meerkat' wording: {msg}",
            );
        }
    }

    #[test]
    fn test_invalid_transition_display() {
        let err = MobError::InvalidTransition {
            from: MobState::Completed,
            to: MobState::Running,
        };
        let msg = format!("{err}");
        assert!(msg.contains("Completed"));
        assert!(msg.contains("Running"));
    }

    #[test]
    fn test_definition_error_display() {
        let err = MobError::DefinitionError(vec![
            Diagnostic {
                code: DiagnosticCode::MissingSkillRef,
                message: "skill 'foo' not found".to_string(),
                location: Some("profiles.worker.skills[0]".to_string()),
                severity: DiagnosticSeverity::Error,
            },
            Diagnostic {
                code: DiagnosticCode::EmptyProfiles,
                message: "no spawnable profiles".to_string(),
                location: Some("profiles".to_string()),
                severity: DiagnosticSeverity::Error,
            },
        ]);
        let msg = format!("{err}");
        assert!(msg.contains("missing_skill_ref"));
        assert!(msg.contains("empty_profiles"));
    }

    #[test]
    fn test_session_error_from() {
        let session_err = meerkat_core::service::SessionError::NotFound {
            id: meerkat_core::types::SessionId::new(),
        };
        let mob_err: MobError = session_err.into();
        assert!(matches!(mob_err, MobError::SessionError(_)));
    }

    #[test]
    fn test_comms_error_from() {
        let send_err = meerkat_core::comms::SendError::PeerNotFound("agent-1".to_string());
        let mob_err: MobError = send_err.into();
        assert!(matches!(mob_err, MobError::CommsError(_)));
    }

    #[test]
    fn test_storage_error() {
        let err = MobError::StorageError(Box::new(std::io::Error::new(
            std::io::ErrorKind::Other,
            "disk full",
        )));
        assert!(format!("{err}").contains("disk full"));
    }

    #[test]
    fn test_all_variants_exist() {
        // Ensures all variants are constructible.
        let _variants: Vec<MobError> = vec![
            MobError::ProfileNotFound(ProfileName::from("p")),
            MobError::MemberNotFound(AgentIdentity::from("m")),
            MobError::MemberAlreadyExists(AgentIdentity::from("m")),
            MobError::NotExternallyAddressable(AgentIdentity::from("m")),
            MobError::InvalidTransition {
                from: MobState::Creating,
                to: MobState::Running,
            },
            MobError::WiringError("w".to_string()),
            MobError::SupervisorRotationIncomplete {
                previous_epoch: 1,
                attempted_epoch: 2,
                attempted_public_peer_id: "peer-next".to_string(),
                rotated_peer_count: 1,
                rollback_succeeded: false,
                pending_authority_recorded: true,
                rollback_error: Some("rollback failed".to_string()),
                reason: "remote failed".to_string(),
            },
            MobError::BridgeCommandRejected {
                cause: BridgeRejectionCause::NotBound,
                reason: "bind required".to_string(),
            },
            MobError::MemberRestoreFailed {
                member_id: AgentIdentity::from("m"),
                session_id: Some(meerkat_core::types::SessionId::new()),
                reason: "restore failed".to_string(),
            },
            MobError::KickoffWaitTimedOut {
                pending_member_ids: vec![AgentIdentity::from("m")],
            },
            MobError::DefinitionError(vec![]),
            MobError::FlowNotFound(FlowId::from("f")),
            MobError::FlowFailed {
                run_id: RunId::new(),
                reason: "r".to_string(),
            },
            MobError::RunNotFound(RunId::new()),
            MobError::RunCanceled(RunId::new()),
            MobError::FlowTurnTimedOut,
            MobError::FrameDepthLimitExceeded {
                loop_id: LoopId::from("loop"),
                max_frame_depth: 1,
                current_depth: 1,
            },
            MobError::FrameAtomicPersistenceUnavailable {
                operation: FrameAtomicOperation::CasGrantNodeSlot,
            },
            MobError::SpecRevisionConflict {
                mob_id: MobId::from("mob"),
                expected: Some(2),
                actual: 3,
            },
            MobError::SchemaValidation {
                step_id: StepId::from("step"),
                message: "invalid".to_string(),
            },
            MobError::InsufficientTargets {
                step_id: StepId::from("step"),
                required: 2,
                available: 1,
            },
            MobError::TopologyViolation {
                from_role: ProfileName::from("lead"),
                to_role: ProfileName::from("worker"),
            },
            MobError::SupervisorEscalation("boom".to_string()),
            MobError::UnsupportedForMode {
                mode: crate::MobRuntimeMode::TurnDriven,
                reason: "autonomous host runtime required".to_string(),
            },
            MobError::ResetBarrier,
            MobError::StorageError(Box::new(std::io::Error::new(
                std::io::ErrorKind::Other,
                "e",
            ))),
            MobError::SessionError(meerkat_core::service::SessionError::PersistenceDisabled),
            MobError::CommsError(meerkat_core::comms::SendError::PeerOffline),
            MobError::StaleFenceToken {
                runtime_id: crate::ids::AgentRuntimeId::initial(crate::ids::AgentIdentity::from(
                    "m",
                )),
                expected: FenceToken::new(1),
                actual: FenceToken::new(0),
            },
            MobError::WorkNotFound(WorkRef::new()),
            MobError::WorkCancellationUnsupported(WorkRef::new()),
            MobError::BridgeSessionNotInLiveAuthority {
                bridge_session_id: "sess-1".to_string(),
            },
            MobError::Internal("i".to_string()),
        ];
    }

    #[test]
    fn failure_class_partitions_missing_target_variants() {
        // The missing-target class is the authority `is_missing_target`
        // delegates to; both must agree on the same variant partition.
        let missing = [
            MobError::MobNotFound(MobId::from("mob")),
            MobError::ProfileNotFound(ProfileName::from("p")),
            MobError::MemberNotFound(AgentIdentity::from("m")),
            MobError::FlowNotFound(FlowId::from("f")),
            MobError::RunNotFound(RunId::new()),
            MobError::WorkNotFound(WorkRef::new()),
        ];
        for err in &missing {
            assert_eq!(
                err.failure_class(),
                MobFailureClass::TargetMissing,
                "{err} should classify as TargetMissing",
            );
            assert!(
                err.is_missing_target(),
                "{err} should report is_missing_target()",
            );
        }
    }

    #[test]
    fn failure_class_maps_non_missing_variants() {
        let cases = [
            (
                MobError::MemberAlreadyExists(AgentIdentity::from("m")),
                MobFailureClass::TargetBusy,
            ),
            (
                MobError::StorageError(Box::new(std::io::Error::new(
                    std::io::ErrorKind::Other,
                    "e",
                ))),
                MobFailureClass::Transport,
            ),
            (MobError::FlowTurnTimedOut, MobFailureClass::Transport),
            (
                MobError::CallbackPending {
                    session_id: meerkat_core::types::SessionId::new(),
                    tool_name: "t".to_string(),
                    args: serde_json::Value::Null,
                },
                MobFailureClass::RuntimeRejected,
            ),
            (
                MobError::Internal("i".to_string()),
                MobFailureClass::Internal,
            ),
            // A variant outside every explicit arm falls through to the
            // mob-rejected default.
            (MobError::ResetBarrier, MobFailureClass::MobRejected),
        ];
        for (err, expected) in &cases {
            assert_eq!(err.failure_class(), *expected, "{err} misclassified");
            assert!(
                !err.is_missing_target(),
                "{err} must not report is_missing_target()",
            );
        }
    }
}