af_workflow/
management.rs

1//! Bounded, owner-scoped queries shared by transports and Agent tools.
2use af_context::{InstanceId, NodeId, RunId, WorkflowDefinitionId};
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6
7/// A keyset page ordered by immutable identity (revision number for revisions).
8#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(default, deny_unknown_fields)]
10pub struct WorkflowPage {
11    /// Exclusive last identity returned by the preceding page.
12    pub after: Option<String>,
13    /// Number of records, between one and 100; default 50.
14    pub limit: u32,
15}
16impl Default for WorkflowPage {
17    fn default() -> Self {
18        Self {
19            after: None,
20            limit: 50,
21        }
22    }
23}
24impl WorkflowPage {
25    /// Reject unbounded requests and malformed cursors before storage access.
26    pub fn validate(&self) -> Result<(), String> {
27        if !(1..=100).contains(&self.limit)
28            || self
29                .after
30                .as_ref()
31                .is_some_and(|s| s.trim().is_empty() || s.len() > 512)
32        {
33            return Err(
34                "workflow page requires limit 1..100 and a nonempty cursor of at most 512 bytes"
35                    .into(),
36            );
37        }
38        Ok(())
39    }
40}
41
42/// Public Workflow read operations. Caller identity is always supplied separately.
43#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
44#[serde(tag = "resource", rename_all = "snake_case", deny_unknown_fields)]
45pub enum WorkflowQuery {
46    /// Definitions authored by this subject (administrators may inspect the tenant).
47    Definitions {
48        /// Include archived definitions; default false.
49        #[serde(default)]
50        include_archived: bool,
51        /// Keyset pagination.
52        #[serde(default)]
53        page: WorkflowPage,
54    },
55    /// Instances owned by this subject, including stopped instances.
56    Instances {
57        /// Include archived instances; default false.
58        #[serde(default)]
59        include_archived: bool,
60        /// Keyset pagination.
61        #[serde(default)]
62        page: WorkflowPage,
63    },
64    /// Immutable revisions of one visible definition.
65    Revisions {
66        /// Definition whose revisions are requested.
67        definition_id: WorkflowDefinitionId,
68        /// Cursor is a decimal revision number.
69        #[serde(default)]
70        page: WorkflowPage,
71    },
72    /// Immutable run facts for one owned instance.
73    Runs {
74        /// Instance whose runs are requested.
75        instance_id: InstanceId,
76        /// Cursor is a run UUID.
77        #[serde(default)]
78        page: WorkflowPage,
79    },
80    /// Immutable step facts for one owned run.
81    Steps {
82        /// Run whose steps are requested.
83        run_id: RunId,
84        /// Cursor is a step UUID.
85        #[serde(default)]
86        page: WorkflowPage,
87    },
88    /// One bounded class of external-action facts for an owned instance.
89    ActionFacts {
90        /// Instance whose action facts are requested.
91        instance_id: InstanceId,
92        /// Fact class. Intent pages omit action_id; the other classes require it.
93        kind: WorkflowActionFactKind,
94        /// Action whose attempts, receipts, or observations are requested.
95        action_id: Option<String>,
96        /// Keyset pagination by action id, attempt number, receipt id, or observation id.
97        #[serde(default)]
98        page: WorkflowPage,
99    },
100}
101impl WorkflowQuery {
102    /// Validate page limits and resource-specific cursor syntax.
103    pub fn validate(&self) -> Result<(), String> {
104        let page = match self {
105            Self::Definitions { page, .. }
106            | Self::Instances { page, .. }
107            | Self::Revisions { page, .. }
108            | Self::Runs { page, .. }
109            | Self::Steps { page, .. }
110            | Self::ActionFacts { page, .. } => page,
111        };
112        page.validate()?;
113        if let Self::Steps { run_id, .. } = self {
114            uuid::Uuid::parse_str(run_id.as_str()).map_err(|_| "invalid workflow run UUID")?;
115        }
116        if let Some(after) = &page.after {
117            match self {
118                Self::Revisions { .. } => {
119                    after
120                        .parse::<i64>()
121                        .ok()
122                        .filter(|n| *n >= 0)
123                        .ok_or("invalid revision cursor")?;
124                }
125                Self::Runs { .. } | Self::Steps { .. } => {
126                    uuid::Uuid::parse_str(after).map_err(|_| "invalid workflow UUID cursor")?;
127                }
128                Self::ActionFacts {
129                    kind: WorkflowActionFactKind::Attempt,
130                    ..
131                } => {
132                    after
133                        .parse::<u32>()
134                        .ok()
135                        .filter(|value| *value > 0)
136                        .ok_or("invalid action attempt cursor")?;
137                }
138                _ => {}
139            }
140        }
141        if let Self::ActionFacts {
142            kind, action_id, ..
143        } = self
144        {
145            let has_action = action_id.as_ref().is_some_and(|id| !id.trim().is_empty());
146            if (*kind == WorkflowActionFactKind::Intent) == has_action {
147                return Err(
148                    "action intent pages omit action_id; attempt/receipt/observation pages require it"
149                        .into(),
150                );
151            }
152        }
153        Ok(())
154    }
155}
156
157/// Independently pageable action fact classes.
158#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
159#[serde(rename_all = "snake_case")]
160pub enum WorkflowActionFactKind {
161    /// Durable external intent and its current lifecycle state.
162    Intent,
163    /// Immutable dispatch-commit attempt.
164    Attempt,
165    /// Immutable provider receipt.
166    Receipt,
167    /// Immutable provider or reconciler observation.
168    Observation,
169}
170
171/// One committed external dispatch attempt.
172#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
173pub struct WorkflowActionAttempt {
174    /// Intent being dispatched.
175    pub action_intent_id: String,
176    /// Stable 1-based attempt number.
177    pub attempt: u32,
178    /// Claim fence at commit.
179    pub action_epoch: i64,
180    /// Stable key sent on every attempt.
181    pub idempotency_key: String,
182    /// Database commit time.
183    pub committed_at: DateTime<Utc>,
184    /// True when reconstructed from a pre-0.7 mutable counter.
185    pub historical: bool,
186}
187
188/// Public action fact. Each query page contains only its requested variant.
189#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
190#[serde(tag = "fact", content = "value", rename_all = "snake_case")]
191pub enum WorkflowActionFact {
192    /// Intent.
193    Intent(crate::ActionIntent),
194    /// Dispatch attempt.
195    Attempt(WorkflowActionAttempt),
196    /// Provider receipt.
197    Receipt(crate::ActionReceipt),
198    /// Provider observation.
199    Observation(crate::ActionObservation),
200}
201
202/// Published definition metadata, independent of immutable revision content.
203#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
204pub struct WorkflowDefinitionSummary {
205    /// Tenant-scoped definition identity.
206    pub definition_id: WorkflowDefinitionId,
207    /// Human-readable name.
208    pub name: String,
209    /// Compare-and-set version of display metadata.
210    pub metadata_version: u64,
211    /// Archived definitions cannot create new instances.
212    pub archived: bool,
213    /// Whether this caller may edit this definition metadata.
214    pub can_manage: bool,
215    /// Time the definition was first published.
216    pub created_at: DateTime<Utc>,
217}
218/// An immutable revision envelope, preserving legacy builtin contract documents.
219#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
220pub struct WorkflowRevisionSummary {
221    /// Definition identity.
222    pub definition_id: WorkflowDefinitionId,
223    /// Published revision number.
224    pub revision: u64,
225    /// Immutable digest.
226    pub content_digest: String,
227    /// Original contract; a Spec revision includes its unchanged `spec` graph.
228    pub contract: Value,
229}
230/// Owned instance status and immutable pins.
231#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
232pub struct WorkflowInstanceSummary {
233    /// Display name, independent of pinned execution content.
234    pub name: String,
235    /// Compare-and-set metadata version.
236    pub metadata_version: u64,
237    /// Whether this stopped instance is archived.
238    pub archived: bool,
239    /// Instance identity.
240    pub instance_id: InstanceId,
241    /// Definition pinned by this instance.
242    pub definition_id: WorkflowDefinitionId,
243    /// Pinned revision number.
244    pub revision: u64,
245    /// Current lifecycle status.
246    pub status: String,
247    /// Instance control mode, separate from lifecycle status; unknown if absent.
248    pub control_mode: String,
249    /// Next durable wakeup time.
250    pub next_run_at: DateTime<Utc>,
251    /// Scheduled occurrence currently being evaluated or awaited.
252    pub schedule_at: DateTime<Utc>,
253    /// Earliest configured activation time.
254    pub starts_at: Option<DateTime<Utc>>,
255    /// Configured expiry time.
256    pub expires_at: Option<DateTime<Utc>>,
257    /// Hard stop after draining dispatched work.
258    pub drain_deadline: Option<DateTime<Utc>>,
259    /// Frozen lifecycle policy for this instance.
260    pub lifecycle: crate::LifecyclePolicy,
261    /// Most recent failure, when present.
262    pub last_error: Option<String>,
263}
264/// Execution facts for a single run.
265#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
266pub struct WorkflowRunSummary {
267    /// Durable run identity.
268    pub run_id: RunId,
269    /// Trigger class.
270    pub trigger_kind: String,
271    /// Current recorded status.
272    pub status: String,
273    /// Stable terminal reason, when recorded.
274    pub exit_reason: Option<String>,
275    /// Start time.
276    pub started_at: DateTime<Utc>,
277    /// Completion time.
278    pub finished_at: Option<DateTime<Utc>>,
279}
280/// One append-only step fact.
281#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
282pub struct WorkflowStepSummary {
283    /// UUID of the recorded step (also the page cursor).
284    pub step_id: af_context::WorkflowStepId,
285    /// Node that produced the fact.
286    pub node_id: NodeId,
287    /// Registered node type.
288    pub node_type: String,
289    /// Recorded outcome.
290    pub status: String,
291    /// Stable exit reason.
292    pub exit_reason: Option<String>,
293    /// Structured diagnostic facts.
294    pub detail: Value,
295    /// Recording time.
296    pub created_at: DateTime<Utc>,
297}
298
299/// Rebuildable operator verdict derived from authoritative workflow facts.
300#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
301#[serde(rename_all = "snake_case")]
302pub enum WorkflowVerdictKind {
303    /// No blocking fact is currently present.
304    Healthy,
305    /// Lifecycle progress or event timeout terminalized the instance.
306    Stalled,
307    /// An ordered source has a missing range.
308    SourceGap,
309    /// The pinned provider circuit or bulkhead blocked work.
310    ProviderBlocked,
311    /// An external action may have committed and requires reconciliation.
312    ActionUncertain,
313    /// Instance reached a non-running lifecycle state for another reason.
314    Terminal,
315}
316/// One bounded query result. `next` is absent when no more rows were observed.
317#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
318#[serde(tag = "resource", rename_all = "snake_case")]
319pub enum WorkflowQueryResult {
320    /// Definition metadata.
321    Definitions {
322        /// Returned records.
323        items: Vec<WorkflowDefinitionSummary>,
324        /// Exclusive next cursor.
325        next: Option<String>,
326    },
327    /// Instance summaries.
328    Instances {
329        /// Returned records.
330        items: Vec<WorkflowInstanceSummary>,
331        /// Exclusive next cursor.
332        next: Option<String>,
333    },
334    /// Full pinned revisions; graph is the original Spec JSON.
335    Revisions {
336        /// Returned records.
337        items: Vec<WorkflowRevisionSummary>,
338        /// Exclusive next cursor.
339        next: Option<String>,
340    },
341    /// Recorded runs.
342    Runs {
343        /// Returned records.
344        items: Vec<WorkflowRunSummary>,
345        /// Exclusive next cursor.
346        next: Option<String>,
347    },
348    /// Recorded steps.
349    Steps {
350        /// Returned records.
351        items: Vec<WorkflowStepSummary>,
352        /// Exclusive next cursor.
353        next: Option<String>,
354    },
355    /// External action facts.
356    ActionFacts {
357        /// Returned records.
358        items: Vec<WorkflowActionFact>,
359        /// Exclusive next cursor.
360        next: Option<String>,
361    },
362}
363
364#[cfg(test)]
365mod tests {
366    use super::*;
367    #[test]
368    fn bounded_query_rejects_unknown_fields_and_invalid_cursors() {
369        for resource in ["definitions", "instances", "revisions", "runs", "steps"] {
370            let mut value = serde_json::json!({"resource": resource, "page": {}, "definition_id":"definition", "instance_id":"instance", "run_id":"00000000-0000-0000-0000-000000000001"});
371            for field in ["definition_id", "instance_id", "run_id"] {
372                if !matches!(
373                    (resource, field),
374                    ("revisions", "definition_id") | ("runs", "instance_id") | ("steps", "run_id")
375                ) {
376                    value.as_object_mut().unwrap().remove(field);
377                }
378            }
379            let parse = |value: &Value| serde_json::from_value::<WorkflowQuery>(value.clone());
380            assert!(parse(&value).unwrap().validate().is_ok());
381            for limit in [0, 101] {
382                value["page"]["limit"] = limit.into();
383                assert!(parse(&value).unwrap().validate().is_err());
384            }
385            value["page"]["limit"] = 100.into();
386            for after in ["".to_owned(), "x".repeat(513)] {
387                value["page"]["after"] = after.into();
388                assert!(parse(&value).unwrap().validate().is_err());
389            }
390            value["page"]["after"] = match resource {
391                "revisions" => "0",
392                "runs" | "steps" => "00000000-0000-0000-0000-000000000001",
393                _ => "last",
394            }
395            .into();
396            assert!(parse(&value).unwrap().validate().is_ok());
397            if matches!(resource, "revisions" | "runs" | "steps") {
398                value["page"]["after"] = "bad".into();
399                assert!(parse(&value).unwrap().validate().is_err());
400            }
401            if resource == "steps" {
402                value["run_id"] = "bad".into();
403                assert!(parse(&value).unwrap().validate().is_err());
404            }
405            value["subject_id"] = "other".into();
406            assert!(parse(&value).is_err());
407        }
408        let intent: WorkflowQuery = serde_json::from_value(serde_json::json!({
409            "resource":"action_facts","instance_id":"instance","kind":"intent","page":{}
410        }))
411        .unwrap();
412        assert!(intent.validate().is_ok());
413        let attempt: WorkflowQuery = serde_json::from_value(serde_json::json!({
414            "resource":"action_facts","instance_id":"instance","kind":"attempt","action_id":"action","page":{"after":"1"}
415        })).unwrap();
416        assert!(attempt.validate().is_ok());
417        for invalid in [
418            serde_json::json!({"resource":"action_facts","instance_id":"instance","kind":"intent","action_id":"action","page":{}}),
419            serde_json::json!({"resource":"action_facts","instance_id":"instance","kind":"receipt","page":{}}),
420            serde_json::json!({"resource":"action_facts","instance_id":"instance","kind":"attempt","action_id":"action","page":{"after":"bad"}}),
421        ] {
422            assert!(serde_json::from_value::<WorkflowQuery>(invalid)
423                .unwrap()
424                .validate()
425                .is_err());
426        }
427    }
428}
af_workflow/management.rs

af_workflow/
management.rs
