af-workflow 0.7.1

//! Bounded, owner-scoped queries shared by transports and Agent tools.
use af_context::{InstanceId, NodeId, RunId, WorkflowDefinitionId};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;

/// A keyset page ordered by immutable identity (revision number for revisions).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct WorkflowPage {
    /// Exclusive last identity returned by the preceding page.
    pub after: Option<String>,
    /// Number of records, between one and 100; default 50.
    pub limit: u32,
}
impl Default for WorkflowPage {
    fn default() -> Self {
        Self {
            after: None,
            limit: 50,
        }
    }
}
impl WorkflowPage {
    /// Reject unbounded requests and malformed cursors before storage access.
    pub fn validate(&self) -> Result<(), String> {
        if !(1..=100).contains(&self.limit)
            || self
                .after
                .as_ref()
                .is_some_and(|s| s.trim().is_empty() || s.len() > 512)
        {
            return Err(
                "workflow page requires limit 1..100 and a nonempty cursor of at most 512 bytes"
                    .into(),
            );
        }
        Ok(())
    }
}

/// Public Workflow read operations. Caller identity is always supplied separately.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "resource", rename_all = "snake_case", deny_unknown_fields)]
pub enum WorkflowQuery {
    /// Definitions authored by this subject (administrators may inspect the tenant).
    Definitions {
        /// Include archived definitions; default false.
        #[serde(default)]
        include_archived: bool,
        /// Keyset pagination.
        #[serde(default)]
        page: WorkflowPage,
    },
    /// Instances owned by this subject, including stopped instances.
    Instances {
        /// Include archived instances; default false.
        #[serde(default)]
        include_archived: bool,
        /// Keyset pagination.
        #[serde(default)]
        page: WorkflowPage,
    },
    /// Immutable revisions of one visible definition.
    Revisions {
        /// Definition whose revisions are requested.
        definition_id: WorkflowDefinitionId,
        /// Cursor is a decimal revision number.
        #[serde(default)]
        page: WorkflowPage,
    },
    /// Immutable run facts for one owned instance.
    Runs {
        /// Instance whose runs are requested.
        instance_id: InstanceId,
        /// Cursor is a run UUID.
        #[serde(default)]
        page: WorkflowPage,
    },
    /// Immutable step facts for one owned run.
    Steps {
        /// Run whose steps are requested.
        run_id: RunId,
        /// Cursor is a step UUID.
        #[serde(default)]
        page: WorkflowPage,
    },
    /// One bounded class of external-action facts for an owned instance.
    ActionFacts {
        /// Instance whose action facts are requested.
        instance_id: InstanceId,
        /// Fact class. Intent pages omit action_id; the other classes require it.
        kind: WorkflowActionFactKind,
        /// Action whose attempts, receipts, or observations are requested.
        action_id: Option<String>,
        /// Keyset pagination by action id, attempt number, receipt id, or observation id.
        #[serde(default)]
        page: WorkflowPage,
    },
}
impl WorkflowQuery {
    /// Validate page limits and resource-specific cursor syntax.
    pub fn validate(&self) -> Result<(), String> {
        let page = match self {
            Self::Definitions { page, .. }
            | Self::Instances { page, .. }
            | Self::Revisions { page, .. }
            | Self::Runs { page, .. }
            | Self::Steps { page, .. }
            | Self::ActionFacts { page, .. } => page,
        };
        page.validate()?;
        if let Self::Steps { run_id, .. } = self {
            uuid::Uuid::parse_str(run_id.as_str()).map_err(|_| "invalid workflow run UUID")?;
        }
        if let Some(after) = &page.after {
            match self {
                Self::Revisions { .. } => {
                    after
                        .parse::<i64>()
                        .ok()
                        .filter(|n| *n >= 0)
                        .ok_or("invalid revision cursor")?;
                }
                Self::Runs { .. } | Self::Steps { .. } => {
                    uuid::Uuid::parse_str(after).map_err(|_| "invalid workflow UUID cursor")?;
                }
                Self::ActionFacts {
                    kind: WorkflowActionFactKind::Attempt,
                    ..
                } => {
                    after
                        .parse::<u32>()
                        .ok()
                        .filter(|value| *value > 0)
                        .ok_or("invalid action attempt cursor")?;
                }
                _ => {}
            }
        }
        if let Self::ActionFacts {
            kind, action_id, ..
        } = self
        {
            let has_action = action_id.as_ref().is_some_and(|id| !id.trim().is_empty());
            if (*kind == WorkflowActionFactKind::Intent) == has_action {
                return Err(
                    "action intent pages omit action_id; attempt/receipt/observation pages require it"
                        .into(),
                );
            }
        }
        Ok(())
    }
}

/// Independently pageable action fact classes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowActionFactKind {
    /// Durable external intent and its current lifecycle state.
    Intent,
    /// Immutable dispatch-commit attempt.
    Attempt,
    /// Immutable provider receipt.
    Receipt,
    /// Immutable provider or reconciler observation.
    Observation,
}

/// One committed external dispatch attempt.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowActionAttempt {
    /// Intent being dispatched.
    pub action_intent_id: String,
    /// Stable 1-based attempt number.
    pub attempt: u32,
    /// Claim fence at commit.
    pub action_epoch: i64,
    /// Stable key sent on every attempt.
    pub idempotency_key: String,
    /// Database commit time.
    pub committed_at: DateTime<Utc>,
    /// True when reconstructed from a pre-0.7 mutable counter.
    pub historical: bool,
}

/// Public action fact. Each query page contains only its requested variant.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "fact", content = "value", rename_all = "snake_case")]
pub enum WorkflowActionFact {
    /// Intent.
    Intent(crate::ActionIntent),
    /// Dispatch attempt.
    Attempt(WorkflowActionAttempt),
    /// Provider receipt.
    Receipt(crate::ActionReceipt),
    /// Provider observation.
    Observation(crate::ActionObservation),
}

/// Published definition metadata, independent of immutable revision content.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowDefinitionSummary {
    /// Tenant-scoped definition identity.
    pub definition_id: WorkflowDefinitionId,
    /// Human-readable name.
    pub name: String,
    /// Compare-and-set version of display metadata.
    pub metadata_version: u64,
    /// Archived definitions cannot create new instances.
    pub archived: bool,
    /// Whether this caller may edit this definition metadata.
    pub can_manage: bool,
    /// Time the definition was first published.
    pub created_at: DateTime<Utc>,
}
/// An immutable revision envelope, preserving legacy builtin contract documents.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowRevisionSummary {
    /// Definition identity.
    pub definition_id: WorkflowDefinitionId,
    /// Published revision number.
    pub revision: u64,
    /// Immutable digest.
    pub content_digest: String,
    /// Original contract; a Spec revision includes its unchanged `spec` graph.
    pub contract: Value,
}
/// Owned instance status and immutable pins.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowInstanceSummary {
    /// Display name, independent of pinned execution content.
    pub name: String,
    /// Compare-and-set metadata version.
    pub metadata_version: u64,
    /// Whether this stopped instance is archived.
    pub archived: bool,
    /// Instance identity.
    pub instance_id: InstanceId,
    /// Definition pinned by this instance.
    pub definition_id: WorkflowDefinitionId,
    /// Pinned revision number.
    pub revision: u64,
    /// Current lifecycle status.
    pub status: String,
    /// Instance control mode, separate from lifecycle status; unknown if absent.
    pub control_mode: String,
    /// Next durable wakeup time.
    pub next_run_at: DateTime<Utc>,
    /// Scheduled occurrence currently being evaluated or awaited.
    pub schedule_at: DateTime<Utc>,
    /// Earliest configured activation time.
    pub starts_at: Option<DateTime<Utc>>,
    /// Configured expiry time.
    pub expires_at: Option<DateTime<Utc>>,
    /// Hard stop after draining dispatched work.
    pub drain_deadline: Option<DateTime<Utc>>,
    /// Frozen lifecycle policy for this instance.
    pub lifecycle: crate::LifecyclePolicy,
    /// Most recent failure, when present.
    pub last_error: Option<String>,
}
/// Execution facts for a single run.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowRunSummary {
    /// Durable run identity.
    pub run_id: RunId,
    /// Trigger class.
    pub trigger_kind: String,
    /// Current recorded status.
    pub status: String,
    /// Stable terminal reason, when recorded.
    pub exit_reason: Option<String>,
    /// Start time.
    pub started_at: DateTime<Utc>,
    /// Completion time.
    pub finished_at: Option<DateTime<Utc>>,
}
/// One append-only step fact.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowStepSummary {
    /// UUID of the recorded step (also the page cursor).
    pub step_id: af_context::WorkflowStepId,
    /// Node that produced the fact.
    pub node_id: NodeId,
    /// Registered node type.
    pub node_type: String,
    /// Recorded outcome.
    pub status: String,
    /// Stable exit reason.
    pub exit_reason: Option<String>,
    /// Structured diagnostic facts.
    pub detail: Value,
    /// Recording time.
    pub created_at: DateTime<Utc>,
}

/// Rebuildable operator verdict derived from authoritative workflow facts.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowVerdictKind {
    /// No blocking fact is currently present.
    Healthy,
    /// Lifecycle progress or event timeout terminalized the instance.
    Stalled,
    /// An ordered source has a missing range.
    SourceGap,
    /// The pinned provider circuit or bulkhead blocked work.
    ProviderBlocked,
    /// An external action may have committed and requires reconciliation.
    ActionUncertain,
    /// Instance reached a non-running lifecycle state for another reason.
    Terminal,
}
/// One bounded query result. `next` is absent when no more rows were observed.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "resource", rename_all = "snake_case")]
pub enum WorkflowQueryResult {
    /// Definition metadata.
    Definitions {
        /// Returned records.
        items: Vec<WorkflowDefinitionSummary>,
        /// Exclusive next cursor.
        next: Option<String>,
    },
    /// Instance summaries.
    Instances {
        /// Returned records.
        items: Vec<WorkflowInstanceSummary>,
        /// Exclusive next cursor.
        next: Option<String>,
    },
    /// Full pinned revisions; graph is the original Spec JSON.
    Revisions {
        /// Returned records.
        items: Vec<WorkflowRevisionSummary>,
        /// Exclusive next cursor.
        next: Option<String>,
    },
    /// Recorded runs.
    Runs {
        /// Returned records.
        items: Vec<WorkflowRunSummary>,
        /// Exclusive next cursor.
        next: Option<String>,
    },
    /// Recorded steps.
    Steps {
        /// Returned records.
        items: Vec<WorkflowStepSummary>,
        /// Exclusive next cursor.
        next: Option<String>,
    },
    /// External action facts.
    ActionFacts {
        /// Returned records.
        items: Vec<WorkflowActionFact>,
        /// Exclusive next cursor.
        next: Option<String>,
    },
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn bounded_query_rejects_unknown_fields_and_invalid_cursors() {
        for resource in ["definitions", "instances", "revisions", "runs", "steps"] {
            let mut value = serde_json::json!({"resource": resource, "page": {}, "definition_id":"definition", "instance_id":"instance", "run_id":"00000000-0000-0000-0000-000000000001"});
            for field in ["definition_id", "instance_id", "run_id"] {
                if !matches!(
                    (resource, field),
                    ("revisions", "definition_id") | ("runs", "instance_id") | ("steps", "run_id")
                ) {
                    value.as_object_mut().unwrap().remove(field);
                }
            }
            let parse = |value: &Value| serde_json::from_value::<WorkflowQuery>(value.clone());
            assert!(parse(&value).unwrap().validate().is_ok());
            for limit in [0, 101] {
                value["page"]["limit"] = limit.into();
                assert!(parse(&value).unwrap().validate().is_err());
            }
            value["page"]["limit"] = 100.into();
            for after in ["".to_owned(), "x".repeat(513)] {
                value["page"]["after"] = after.into();
                assert!(parse(&value).unwrap().validate().is_err());
            }
            value["page"]["after"] = match resource {
                "revisions" => "0",
                "runs" | "steps" => "00000000-0000-0000-0000-000000000001",
                _ => "last",
            }
            .into();
            assert!(parse(&value).unwrap().validate().is_ok());
            if matches!(resource, "revisions" | "runs" | "steps") {
                value["page"]["after"] = "bad".into();
                assert!(parse(&value).unwrap().validate().is_err());
            }
            if resource == "steps" {
                value["run_id"] = "bad".into();
                assert!(parse(&value).unwrap().validate().is_err());
            }
            value["subject_id"] = "other".into();
            assert!(parse(&value).is_err());
        }
        let intent: WorkflowQuery = serde_json::from_value(serde_json::json!({
            "resource":"action_facts","instance_id":"instance","kind":"intent","page":{}
        }))
        .unwrap();
        assert!(intent.validate().is_ok());
        let attempt: WorkflowQuery = serde_json::from_value(serde_json::json!({
            "resource":"action_facts","instance_id":"instance","kind":"attempt","action_id":"action","page":{"after":"1"}
        })).unwrap();
        assert!(attempt.validate().is_ok());
        for invalid in [
            serde_json::json!({"resource":"action_facts","instance_id":"instance","kind":"intent","action_id":"action","page":{}}),
            serde_json::json!({"resource":"action_facts","instance_id":"instance","kind":"receipt","page":{}}),
            serde_json::json!({"resource":"action_facts","instance_id":"instance","kind":"attempt","action_id":"action","page":{"after":"bad"}}),
        ] {
            assert!(serde_json::from_value::<WorkflowQuery>(invalid)
                .unwrap()
                .validate()
                .is_err());
        }
    }
}