frame-conv 0.2.0

//! The typed workflow observation vocabulary: progress projections of
//! every nonterminal history event and the closed terminal outcome set.
//!
//! Projection law (R2): every recorded aion event this pin can publish
//! has exactly one typed projection here, each preserving the recorded
//! sequence identity — nothing is dropped or flattened. The projection
//! match in `observe.rs` is EXHAUSTIVE over the pinned `aion-core`
//! `=0.10.0` event enum, so an upstream pin bump that adds a variant
//! fails loudly at compile time and forces this vocabulary to be trued —
//! the strongest available form of "unknown events fail typed and loud".
//!
//! Semantic time only (design constraint 5): wall-clock timestamps
//! recorded in history (timer fire deadlines, record times) are engine
//! bookkeeping and are deliberately NOT projected; sequence identity is
//! the only order this surface speaks.

use aion_core::{ActivityError, Payload, WorkflowError};
use serde::de::DeserializeOwned;

use super::id::{WorkflowConversationId, WorkflowRunId, WorkflowStepId};

/// One observed item from a workflow conversation: nonterminal progress
/// or a member of the closed terminal set.
#[derive(Debug)]
pub enum WorkflowItem {
    /// A typed projection of one nonterminal history event.
    Progress(WorkflowProgress),
    /// A member of the closed terminal outcome set.
    Terminal(WorkflowTerminal),
}

/// A typed nonterminal progress item, preserving the recorded sequence.
#[derive(Debug)]
pub struct WorkflowProgress {
    /// The event's recorded per-workflow sequence number.
    pub seq: u64,
    /// What happened, in conversation vocabulary.
    pub kind: WorkflowProgressKind,
}

/// The closed nonterminal progress vocabulary — one variant per pinned
/// nonterminal history event family.
#[derive(Debug)]
pub enum WorkflowProgressKind {
    /// The run opened (or a continue-as-new successor run opened, in
    /// which case `continued_from` names the run it continues).
    Opened {
        /// Workflow kind the run executes.
        workflow_kind: String,
        /// The run that opened.
        run: WorkflowRunId,
        /// The prior run this one continues, on a continue-as-new chain.
        continued_from: Option<WorkflowRunId>,
    },
    /// A failed or cancelled run was reopened and returned to live.
    Reopened {
        /// The run being reopened.
        run: WorkflowRunId,
        /// Steps whose recorded failures are superseded for re-dispatch.
        redispatched_steps: Vec<WorkflowStepId>,
    },
    /// An operator paused the run — NONTERMINAL: durable record paths
    /// keep recording and the run can still complete, fail, or resume.
    Paused {
        /// Operator-supplied reason, when one was given.
        reason: Option<String>,
    },
    /// An operator resumed a paused run.
    Resumed,
    /// Visibility attributes changed; the names are news, the values
    /// live in aion's visibility store.
    AttributesNoted {
        /// Names of the updated attributes.
        attributes: Vec<String>,
    },
    /// Workflow code scheduled a step.
    StepScheduled {
        /// The step's deterministic identity.
        step: WorkflowStepId,
        /// Step kind selected by workflow code.
        step_kind: String,
        /// Opaque step input.
        input: WorkflowPayload,
    },
    /// A worker began executing a step attempt.
    StepStarted {
        /// The step being executed.
        step: WorkflowStepId,
        /// One-based attempt number (`0` = recorded before attempts were
        /// tracked upstream).
        attempt: u32,
    },
    /// A step attempt completed.
    StepCompleted {
        /// The step that completed.
        step: WorkflowStepId,
        /// One-based attempt number that produced the result.
        attempt: u32,
        /// Opaque step result.
        result: WorkflowPayload,
    },
    /// A step attempt failed, retryably or terminally.
    StepFailed {
        /// The step whose attempt failed.
        step: WorkflowStepId,
        /// One-based attempt number that failed.
        attempt: u32,
        /// The classified failure.
        failure: WorkflowStepFailure,
    },
    /// A step attempt was cancelled.
    StepCancelled {
        /// The step that was cancelled.
        step: WorkflowStepId,
        /// One-based attempt number that was cancelled.
        attempt: u32,
    },
    /// A durable timer was armed. Its wall-clock deadline is engine
    /// bookkeeping and is not projected (semantic time only).
    TimerArmed {
        /// The timer's recorded name.
        timer: String,
    },
    /// A durable timer fired.
    TimerFired {
        /// The timer that fired.
        timer: String,
    },
    /// A durable timer was retired without firing. `permanent` is true
    /// when workflow code retired it (never resurrected); false when the
    /// engine retired it tearing down a cancelled run (re-armed on
    /// reopen).
    TimerRetired {
        /// The timer that was retired.
        timer: String,
        /// Whether the retirement survives reopen.
        permanent: bool,
    },
    /// A bounded operation settled: either the operation won or its
    /// deadline did.
    TimeoutSettled {
        /// The timer that bounded the operation.
        timer: String,
        /// True when the deadline fired before the operation completed.
        timed_out: bool,
        /// The operation's result, when it completed.
        result: Option<WorkflowPayload>,
    },
    /// A contribution (aion signal) was delivered to this workflow.
    ContributionReceived {
        /// Contribution name selected by the sender.
        name: String,
        /// Opaque contribution payload.
        payload: WorkflowPayload,
    },
    /// This workflow sent a contribution to another workflow.
    ContributionForwarded {
        /// The receiving conversation.
        target: WorkflowConversationId,
        /// Contribution name selected by workflow code.
        name: String,
    },
    /// A child workflow conversation was opened by this run.
    ChildOpened {
        /// The child conversation's identity.
        child: WorkflowConversationId,
        /// Workflow kind the child executes.
        workflow_kind: String,
    },
    /// A child workflow conversation completed.
    ChildCompleted {
        /// The child that completed.
        child: WorkflowConversationId,
        /// Opaque child result.
        result: WorkflowPayload,
    },
    /// A child workflow conversation failed terminally.
    ChildFailed {
        /// The child that failed.
        child: WorkflowConversationId,
        /// The terminal child failure.
        failure: WorkflowFailure,
    },
    /// A child workflow conversation was cancelled.
    ChildCancelled {
        /// The child that was cancelled.
        child: WorkflowConversationId,
    },
    /// A schedule resource recorded lifecycle news in this history.
    ScheduleNoted {
        /// The schedule's recorded name.
        schedule: String,
        /// What the schedule did.
        note: ScheduleNote,
    },
}

/// Schedule lifecycle news projected from recorded history.
#[derive(Debug)]
pub enum ScheduleNote {
    /// The schedule was created.
    Created,
    /// The schedule's configuration was updated.
    Updated,
    /// The schedule was paused.
    Paused,
    /// The schedule was resumed.
    Resumed,
    /// The schedule was deleted.
    Deleted,
    /// A schedule tick opened a workflow run.
    Triggered {
        /// The conversation the tick opened.
        conversation: WorkflowConversationId,
        /// The run the tick opened.
        run: WorkflowRunId,
    },
}

/// The closed terminal outcome set (R2): exactly these five, each
/// preserving the recorded sequence of its terminal event.
///
/// Two contracts a consumer must not guess (both characterized and
/// pinned at the published engine bytes, `pattern_workflow.rs`):
///
/// - **The completed result is an outcome-tagged envelope.** A workflow
///   that routes an outcome records `{"outcome": "<outcome name>",
///   "payload": <constructed payload>}` as its result — decode
///   [`Completed::result`](Self::Completed) against that envelope, never
///   against the bare payload type.
/// - **A replayed terminal is HISTORY unless it is the frontier.** A
///   reopened run's replay from an early cursor re-surfaces the
///   superseded terminal (the Cancelled/Failed that was reopened past)
///   before the Reopened progress item; supersession is the engine's
///   cursor law, not the stream's. A consumer treating every terminal as
///   final will wrongly end at the superseded one — the run's live fate
///   is decided only by the LAST lifecycle item observed.
#[derive(Debug)]
pub enum WorkflowTerminal {
    /// The run completed with a result.
    Completed {
        /// Recorded sequence of the terminal event.
        seq: u64,
        /// Opaque workflow result — the outcome-tagged envelope
        /// `{"outcome": ..., "payload": ...}` (see the enum docs).
        result: WorkflowPayload,
    },
    /// The run failed terminally.
    Failed {
        /// Recorded sequence of the terminal event.
        seq: u64,
        /// The terminal failure.
        failure: WorkflowFailure,
    },
    /// The run was cancelled.
    Cancelled {
        /// Recorded sequence of the terminal event.
        seq: u64,
        /// Human-readable cancellation reason.
        reason: String,
    },
    /// The run timed out.
    TimedOut {
        /// Recorded sequence of the terminal event.
        seq: u64,
        /// Descriptor identifying the timeout that elapsed.
        timeout: String,
    },
    /// The run continued as a new run of the SAME conversation. The
    /// successor announces itself on this same observation as a
    /// [`WorkflowProgressKind::Opened`] item whose `continued_from`
    /// names `continued_run` — the handle's continue-as-new behavior is
    /// to keep observing, never to re-resolve anything.
    ContinuedAsNew {
        /// Recorded sequence of the terminal event.
        seq: u64,
        /// The run that ended by continuing.
        continued_run: WorkflowRunId,
        /// Opaque input carried into the successor run.
        next_input: WorkflowPayload,
        /// Workflow kind override for the successor, when migrating.
        next_kind: Option<String>,
    },
}

/// An opaque typed payload crossing the workflow seam (input, result,
/// contribution body). Decode is explicit and typed — never implicit.
#[derive(Debug, Clone, PartialEq)]
pub struct WorkflowPayload(Payload);

impl WorkflowPayload {
    /// Decodes the payload as JSON into a typed value.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowCallError::InvalidArgument`] when the payload is
    /// not JSON or does not match `T`.
    pub fn decode<T: DeserializeOwned>(&self) -> Result<T, super::WorkflowCallError> {
        let value =
            self.0
                .to_json()
                .map_err(|error| super::WorkflowCallError::InvalidArgument {
                    message: format!("payload is not JSON: {error}"),
                })?;
        serde_json::from_value(value).map_err(|error| super::WorkflowCallError::InvalidArgument {
            message: format!("payload does not match the requested type: {error}"),
        })
    }

    /// Returns the raw payload bytes.
    #[must_use]
    pub fn bytes(&self) -> &[u8] {
        self.0.bytes()
    }

    pub(crate) fn from_core(payload: Payload) -> Self {
        Self(payload)
    }
}

/// A terminal workflow failure, in conversation vocabulary.
#[derive(Debug, Clone, PartialEq)]
pub struct WorkflowFailure(WorkflowError);

impl WorkflowFailure {
    /// Human-readable failure message.
    #[must_use]
    pub fn message(&self) -> &str {
        &self.0.message
    }

    /// Structured failure details, when the workflow recorded any.
    #[must_use]
    pub fn details(&self) -> Option<WorkflowPayload> {
        self.0.details.clone().map(WorkflowPayload::from_core)
    }

    pub(crate) fn from_core(error: WorkflowError) -> Self {
        Self(error)
    }
}

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

/// A classified step-attempt failure, in conversation vocabulary.
#[derive(Debug, Clone, PartialEq)]
pub struct WorkflowStepFailure(ActivityError);

impl WorkflowStepFailure {
    /// Human-readable failure message.
    #[must_use]
    pub fn message(&self) -> &str {
        &self.0.message
    }

    /// Whether the engine may retry the failed attempt.
    #[must_use]
    pub fn is_retryable(&self) -> bool {
        self.0.is_retryable()
    }

    pub(crate) fn from_core(error: ActivityError) -> Self {
        Self(error)
    }
}

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

/// The run's projected lifecycle phase — a pure projection of committed
/// history, never an independently stored field.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WorkflowPhase {
    /// No terminal lifecycle event is recorded.
    Running,
    /// The run completed.
    Completed,
    /// The run failed terminally.
    Failed,
    /// The run was cancelled.
    Cancelled,
    /// The run timed out.
    TimedOut,
    /// The run continued as a new run.
    ContinuedAsNew,
    /// The run is operator-paused — NONTERMINAL.
    Paused,
}

impl WorkflowPhase {
    /// Whether this phase is a member of the closed terminal set.
    #[must_use]
    pub const fn is_terminal(self) -> bool {
        !matches!(self, Self::Running | Self::Paused)
    }
}

/// The outcome of reopening a terminal reopenable run.
#[derive(Debug)]
pub struct ReopenedRun {
    /// The run the reopened execution continues.
    pub run: WorkflowRunId,
    /// The run's projected phase after reopen (Running).
    pub phase: WorkflowPhase,
}

/// A point-in-time description of the conversation, for EXPLICIT
/// inspection only.
///
/// This surface is never a polling authority for progress: change is
/// observed exclusively through [`super::WorkflowObservation`], which
/// parks until the substrate publishes. Re-arming inspection to watch
/// for change is exactly the shape the no-poll tripwire exists to
/// refuse.
#[derive(Debug)]
pub struct WorkflowInspection {
    /// The run's projected phase, scanned from committed history.
    pub phase: WorkflowPhase,
    /// Number of recorded history events at inspection time.
    pub recorded_events: u64,
}