app_kit/workflow/history.rs
1use serde::{Deserialize, Deserializer, Serialize, Serializer};
2use serde_json::{Map, Value};
3
4use crate::workflow::run::RunRecord;
5use crate::workflow::{RunId, RunSelector, RunStatus, WorkflowId, WorkflowState};
6use crate::{AppError, WorkspaceId};
7
8/// The result of one successfully committed action, distinct from user state.
9///
10/// See [the execution and history workflow](crate::workflow#execute-observe-restore-and-delete).
11#[non_exhaustive]
12#[derive(Clone, Deserialize, PartialEq, Serialize)]
13pub struct ActionOutcome {
14 /// Node that produced this result.
15 pub node: String,
16 /// Semantic action ordinal, independent of control-checkpoint sequence.
17 pub ordinal: u64,
18 /// Exact result fields, distinct from durable workflow state.
19 pub result: Map<String, Value>,
20}
21
22/// The boundary represented by one retained checkpoint.
23pub enum CheckpointKind {
24 /// Initial state admitted before any action executes.
25 Initial,
26 /// State-preserving control transition.
27 Control,
28 /// Successful action commit.
29 Action,
30 /// A failed action observation; committed state and position remain unchanged.
31 Failed {
32 /// Failed activation, or the canonical representative of a failed BSP wave.
33 node: String,
34 /// Proposed action ordinal of that failed activation.
35 ordinal: u64,
36 },
37 /// Terminal workflow transition.
38 Terminal,
39}
40
41/// A committed workflow boundary and its original run identity.
42///
43/// See [the execution and history workflow](crate::workflow#execute-observe-restore-and-delete).
44#[non_exhaustive]
45pub struct Checkpoint {
46 /// Run containing this committed boundary.
47 pub run: RunId,
48 /// Exact admitted occurrence containing this boundary.
49 pub occurrence: String,
50 /// Monotonic checkpoint sequence within the run.
51 pub sequence: u64,
52 /// Kind of committed boundary.
53 pub kind: CheckpointKind,
54 /// Logical workflow identity.
55 pub workflow: WorkflowId,
56 /// Exact workflow revision used by this run.
57 pub revision: String,
58 /// Last successfully committed action ordinal, if any.
59 pub action_ordinal: Option<u64>,
60 /// Last successfully committed action position, if any.
61 pub position: Option<String>,
62 /// Immutable committed user state.
63 pub state: WorkflowState,
64 /// Durable lifecycle at this boundary. Historical reads do not reconstruct
65 /// actionable targets; a running boundary may contain a pending control.
66 pub status: RunStatus,
67 /// Results committed at this boundary; control transitions invent none.
68 pub outcomes: Vec<ActionOutcome>,
69 /// Source boundary when this run was restored into a new identity.
70 pub lineage: Option<RunLineage>,
71}
72
73/// The source boundary from which a fresh run was restored.
74///
75/// See [the execution and history workflow](crate::workflow#execute-observe-restore-and-delete).
76#[non_exhaustive]
77pub struct RunLineage {
78 /// Original run; restoring does not overwrite this identity.
79 pub source: RunId,
80 /// Exact occurrence of the selected source boundary.
81 pub source_occurrence: String,
82 /// Selected committed source checkpoint.
83 pub checkpoint: u64,
84 /// Selected action ordinal, distinct from the source's last committed ordinal.
85 ///
86 /// An explicit action selector preserves its ordinal. Other selectors use
87 /// a failed observation's attempted ordinal or the last committed ordinal.
88 pub action_ordinal: Option<u64>,
89}
90
91/// One scoped retained-history selection, independent of storage revisions.
92///
93/// Its serialized form uses `kind` and, for sequence/action selections, `value`.
94#[derive(Clone, Deserialize, PartialEq, Serialize)]
95#[serde(
96 tag = "kind",
97 content = "value",
98 rename_all = "snake_case",
99 deny_unknown_fields
100)]
101pub enum CheckpointSelector {
102 /// The latest eligible observation in the selected operation.
103 ///
104 /// Run checkpoint reads use the current App checkpoint. Restore and child
105 /// initializers use the latest saved successful or failed action.
106 Latest,
107 /// One exact immutable checkpoint sequence.
108 Sequence(u64),
109 /// The latest successful or failed observation of this action ordinal.
110 Action(u64),
111}
112
113/// The saved execution boundary used when restoring a fresh run.
114///
115/// Serializes as `after_completed`, `at_saved`, or `entrypoint`.
116#[derive(Clone, Deserialize, PartialEq, Serialize)]
117#[serde(rename_all = "snake_case")]
118pub enum ResumePosition {
119 /// Continue after the selected successful action or complete BSP wave.
120 AfterCompleted,
121 /// Retry the saved action or complete uncommitted BSP wave.
122 AtSaved,
123 /// Restart from the definition's entrypoint without prior-action selection.
124 Entrypoint,
125}
126
127/// Selects and copies clean retained state into a new, parked run.
128///
129/// A fallback is used only when the source or selected checkpoint is absent; it
130/// does not silently replace uncertainty or receive overrides intended for loaded
131/// state. A successful restore copies clean state into a fresh paused run, so
132/// later source deletion cannot strand the copy.
133///
134/// See [the execution and history workflow](crate::workflow#execute-observe-restore-and-delete).
135pub struct RestoreOptions {
136 /// Original run, which is never overwritten by restore.
137 pub source: RunSelector,
138 /// Exact source selection, or the latest saved successful or failed action.
139 pub checkpoint: CheckpointSelector,
140 /// State overrides validated before admission of the copy.
141 pub overrides: WorkflowState,
142 /// Where explicit execution of the copied run begins.
143 pub position: ResumePosition,
144 /// Explicit initial state used only when the requested checkpoint is absent.
145 pub fallback: Option<WorkflowState>,
146}
147
148/// One bounded page of immutable checkpoints in ascending sequence order.
149///
150/// See [the execution and history workflow](crate::workflow#execute-observe-restore-and-delete).
151#[non_exhaustive]
152pub struct HistoryPage {
153 /// Complete retained checkpoints from the captured head cutoff.
154 pub checkpoints: Vec<Checkpoint>,
155 /// Continues the same cutoff and scope; absent when that history is exhausted.
156 pub next: Option<HistoryCursor>,
157}
158
159/// An opaque continuation of a captured run-history view.
160///
161/// Serialization preserves its App generation, run creation scope and committed
162/// head cutoff. It is not a native storage cursor and cannot cross run scopes.
163///
164/// See [the execution and history workflow](crate::workflow#execute-observe-restore-and-delete).
165#[derive(Clone, Debug)]
166pub struct HistoryCursor {
167 workspace: WorkspaceId,
168 workflow: WorkflowId,
169 run: RunId,
170 generation: u64,
171 creation: String,
172 cutoff: u64,
173 after: u64,
174}
175
176impl HistoryCursor {
177 pub(crate) fn issue(
178 head: &RunRecord,
179 generation: u64,
180 creation: &str,
181 after: u64,
182 ) -> Result<Self, AppError> {
183 Self {
184 workspace: head.workspace_id.clone(),
185 workflow: head.workflow_id.clone(),
186 run: head.id.clone(),
187 generation,
188 creation: creation.to_owned(),
189 cutoff: head.checkpoint.sequence,
190 after,
191 }
192 .validated()
193 }
194
195 pub(crate) fn continue_from(
196 &self,
197 head: &RunRecord,
198 generation: u64,
199 creation: &str,
200 ) -> Result<(u64, u64), AppError> {
201 if self.workspace != head.workspace_id
202 || self.workflow != head.workflow_id
203 || self.run != head.id
204 || self.generation != generation
205 || self.creation != creation
206 || self.cutoff > head.checkpoint.sequence
207 {
208 return Err(AppError::InvalidInput(
209 "history cursor scope is invalid".into(),
210 ));
211 }
212 Ok((self.cutoff, self.after))
213 }
214
215 pub(crate) fn advance_to(&self, after: u64) -> Result<Self, AppError> {
216 if after <= self.after {
217 return Err(AppError::Integration("history page did not advance".into()));
218 }
219 Self {
220 after,
221 ..self.clone()
222 }
223 .validated()
224 }
225
226 fn validated(self) -> Result<Self, AppError> {
227 if self.generation == 0
228 || self.creation.len() != 32
229 || !self
230 .creation
231 .bytes()
232 .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
233 || self.after >= self.cutoff
234 {
235 return Err(AppError::InvalidInput("invalid history cursor".into()));
236 }
237 Ok(self)
238 }
239}
240
241impl Serialize for HistoryCursor {
242 fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
243 (
244 1u8,
245 &self.workspace,
246 &self.workflow,
247 &self.run,
248 self.generation,
249 &self.creation,
250 self.cutoff,
251 self.after,
252 )
253 .serialize(serializer)
254 }
255}
256
257impl<'de> Deserialize<'de> for HistoryCursor {
258 fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
259 let (version, workspace, workflow, run, generation, creation, cutoff, after) =
260 <(u8, WorkspaceId, WorkflowId, RunId, u64, String, u64, u64)>::deserialize(
261 deserializer,
262 )?;
263 if version != 1 {
264 return Err(serde::de::Error::custom("invalid history cursor"));
265 }
266 Self {
267 workspace,
268 workflow,
269 run,
270 generation,
271 creation,
272 cutoff,
273 after,
274 }
275 .validated()
276 .map_err(serde::de::Error::custom)
277 }
278}