app_kit/workflow/lifecycle.rs
1use std::future::{Future, poll_fn};
2use std::panic::{AssertUnwindSafe, catch_unwind};
3use std::time::{Duration, SystemTime, UNIX_EPOCH};
4
5use futures_util::FutureExt;
6use serde_json::{Map, Value, json};
7
8use crate::workflow::definition::JsonBoundary;
9use crate::workflow::{
10 ActionContext, ActionMode, ActionOutcome, Checkpoint, GraphDescription, RunId, RunLineage,
11 RunSnapshot, WorkflowFailure, WorkflowState,
12};
13use crate::{AppError, WorkspaceId};
14
15const MAX_LIFECYCLE_OBSERVATION: Duration = Duration::from_secs(30);
16
17/// Caller-owned lifecycle observation without a transport or exporter owner.
18///
19/// Both methods default to success. Asynchronous execution calls the synchronous
20/// method before the asynchronous method; synchronous execution calls only the
21/// synchronous method. Events borrow their authoritative data for the duration
22/// of the call. Implementations own any data they choose to retain afterward.
23///
24/// # Example
25///
26/// ```
27/// use app_kit::{AppError, workflow::{LifecycleEvent, LifecycleHooks}};
28/// use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
29/// struct CountEvents(Arc<AtomicUsize>);
30/// impl LifecycleHooks for CountEvents {
31/// fn observe(&self, event: &LifecycleEvent<'_>) -> Result<(), AppError> {
32/// if matches!(event, LifecycleEvent::ExecuteStarted { .. }) {
33/// self.0.fetch_add(1, Ordering::SeqCst);
34/// }
35/// Ok(())
36/// }
37/// fn observe_async<'a>(&'a self, event: &'a LifecycleEvent<'a>)
38/// -> impl std::future::Future<Output = Result<(), AppError>> + Send + 'a {
39/// async move {
40/// if matches!(event, LifecycleEvent::ExecuteStarted { .. }) {
41/// self.0.fetch_add(10, Ordering::SeqCst);
42/// }
43/// Ok(())
44/// }
45/// }
46/// }
47/// let run = "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?;
48/// let event = LifecycleEvent::ExecuteStarted {
49/// run: &run, occurrence: "00000000000040008000000000000001",
50/// };
51/// let count = Arc::new(AtomicUsize::new(0));
52/// let hooks = CountEvents(Arc::clone(&count));
53/// hooks.observe(&event)?;
54/// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
55/// runtime.block_on(hooks.observe_async(&event))?;
56/// assert_eq!(count.load(Ordering::SeqCst), 11);
57/// # Ok::<(), Box<dyn std::error::Error>>(())
58/// ```
59pub trait LifecycleHooks: Send + Sync + 'static {
60 /// Observes one event synchronously; the default has no effect.
61 ///
62 /// # Errors
63 ///
64 /// Return an error to report functional observation failure. App preserves an
65 /// earlier execution failure as the structured cause of `WorkflowFailure::Hook`;
66 /// a post-commit hook error does not roll back an accepted native commit.
67 ///
68 /// See [the complete lifecycle hook implementation](LifecycleHooks).
69 fn observe(&self, _event: &LifecycleEvent<'_>) -> Result<(), AppError> {
70 Ok(())
71 }
72
73 /// Observes one event asynchronously; the default has no effect.
74 ///
75 /// # Errors
76 ///
77 /// Return an error to report functional observation failure. App preserves an
78 /// earlier execution failure as the structured cause of `WorkflowFailure::Hook`;
79 /// a post-commit hook error does not roll back an accepted native commit.
80 ///
81 /// See [the complete lifecycle hook implementation](LifecycleHooks).
82 fn observe_async<'a>(
83 &'a self,
84 _event: &'a LifecycleEvent<'a>,
85 ) -> impl Future<Output = Result<(), AppError>> + Send + 'a {
86 async { Ok(()) }
87 }
88}
89
90/// An exhaustive borrowed view of one application lifecycle event.
91pub enum LifecycleEvent<'a> {
92 /// Registers a run and its graph with optional child or restore lineage.
93 Registered {
94 /// Workspace containing the run.
95 workspace: &'a WorkspaceId,
96 /// Registered run identity.
97 run: &'a RunId,
98 /// Exact admitted execution occurrence.
99 occurrence: &'a str,
100 /// Graph used by the registered run.
101 graph: &'a GraphDescription,
102 /// Parent action when this is an admitted child.
103 parent: Option<&'a ActionContext>,
104 /// Original boundary when this is a restored run.
105 lineage: Option<&'a RunLineage>,
106 },
107 /// Finishes an admitted child under its exact parent action.
108 ChildFinished {
109 /// Parent action that admitted this child.
110 parent: &'a ActionContext,
111 /// Authoritative child snapshot.
112 snapshot: &'a RunSnapshot,
113 },
114 /// Begins one execution session.
115 ExecuteStarted {
116 /// Run being advanced.
117 run: &'a RunId,
118 /// Exact admitted execution occurrence.
119 occurrence: &'a str,
120 },
121 /// Finishes one execution session.
122 ExecuteFinished {
123 /// Authoritative run snapshot at session finalization.
124 snapshot: &'a RunSnapshot,
125 /// Execution failure, when present.
126 failure: Option<&'a WorkflowFailure>,
127 },
128 /// Begins an action attempt against its committed base and resolved inputs.
129 StepStarted {
130 /// Identity and attempted action ordinal.
131 context: &'a ActionContext,
132 /// Unchanged committed base state.
133 state: &'a WorkflowState,
134 /// Resolved transient action inputs.
135 inputs: &'a Map<String, Value>,
136 },
137 /// Finalizes an attempt, including rollback with no completed outcome.
138 StepFinished {
139 /// Identity and attempted ordinal, including failed attempts.
140 context: &'a ActionContext,
141 /// Committed result, exact failure, or no outcome on EOF, abort or Drop.
142 outcome: Option<Result<&'a ActionOutcome, &'a WorkflowFailure>>,
143 /// Committed boundary; failure retains the unchanged prior boundary.
144 checkpoint: &'a Checkpoint,
145 },
146 /// Opens partial-result observation for one action or addressed span.
147 StreamStarted {
148 /// Action owning this stream.
149 context: &'a ActionContext,
150 /// Addressed span, or the action root.
151 span: Option<&'a str>,
152 },
153 /// Observes one transient partial result.
154 StreamItem {
155 /// Action owning this stream.
156 context: &'a ActionContext,
157 /// Addressed span, or the action root.
158 span: Option<&'a str>,
159 /// Zero-based index within this stream.
160 index: usize,
161 /// Transient result fields; these do not commit state.
162 result: &'a Map<String, Value>,
163 },
164 /// Closes partial-result observation exactly once.
165 StreamFinished {
166 /// Action owning this stream.
167 context: &'a ActionContext,
168 /// Addressed span, or the action root.
169 span: Option<&'a str>,
170 /// Number of observed partial items.
171 items: usize,
172 /// Failure that closed observation, if any.
173 failure: Option<&'a WorkflowFailure>,
174 },
175 /// Opens an explicitly addressed span within one action attempt.
176 SpanStarted {
177 /// Action owning the span.
178 context: &'a ActionContext,
179 /// Unique open span identity within this attempt.
180 id: &'a str,
181 /// Display name; repeated names are permitted.
182 name: &'a str,
183 /// Open parent span, or the action root.
184 parent: Option<&'a str>,
185 /// Dependency span identities in caller-authored order.
186 dependencies: &'a [String],
187 },
188 /// Adds an attribute to an open span or the action root.
189 SpanAttribute {
190 /// Action owning the attribute.
191 context: &'a ActionContext,
192 /// Open span, or the action root.
193 span: Option<&'a str>,
194 /// Attribute name.
195 key: &'a str,
196 /// Attribute value.
197 value: &'a Value,
198 /// Caller tags whose values are strings.
199 tags: &'a Map<String, Value>,
200 },
201 /// Closes an addressed span.
202 SpanFinished {
203 /// Action owning the span.
204 context: &'a ActionContext,
205 /// Exact open span identity.
206 id: &'a str,
207 },
208}
209
210/// Default lifecycle implementation with no observation effects.
211///
212/// # Example
213///
214/// ```
215/// use app_kit::workflow::{LifecycleEvent, LifecycleHooks, NoHooks};
216/// let run = "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?;
217/// let event = LifecycleEvent::ExecuteStarted {
218/// run: &run, occurrence: "00000000000040008000000000000001",
219/// };
220/// NoHooks.observe(&event)?;
221/// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
222/// runtime.block_on(NoHooks.observe_async(&event))?;
223/// # Ok::<(), Box<dyn std::error::Error>>(())
224/// ```
225#[derive(Default)]
226pub struct NoHooks;
227
228impl LifecycleHooks for NoHooks {}
229
230impl LifecycleEvent<'_> {
231 pub(crate) fn deadline(activation: Option<tokio::time::Instant>) -> tokio::time::Instant {
232 let ceiling = tokio::time::Instant::now() + MAX_LIFECYCLE_OBSERVATION;
233 activation.map_or(ceiling, |deadline| deadline.min(ceiling))
234 }
235
236 pub(crate) fn deliver_sync<H: LifecycleHooks>(&self, hooks: &H) -> Result<(), WorkflowFailure> {
237 catch_unwind(AssertUnwindSafe(|| hooks.observe(self)))
238 .map_err(|_| WorkflowFailure::Execution("lifecycle callback panicked".into()))?
239 .map_err(WorkflowFailure::hook)
240 }
241
242 pub(crate) async fn deliver<H: LifecycleHooks>(
243 &self,
244 hooks: &H,
245 mode: ActionMode,
246 deadline: tokio::time::Instant,
247 ) -> Result<(), WorkflowFailure> {
248 if mode == ActionMode::Both {
249 return Err(WorkflowFailure::Execution(
250 "lifecycle requires one actual mode".into(),
251 ));
252 }
253 self.deliver_sync(hooks)?;
254 if tokio::time::Instant::now() >= deadline {
255 return Err(WorkflowFailure::Timeout(
256 "lifecycle observation expired".into(),
257 ));
258 }
259 if mode == ActionMode::Synchronous {
260 return Ok(());
261 }
262 let source = catch_unwind(AssertUnwindSafe(|| hooks.observe_async(self)))
263 .map_err(|_| WorkflowFailure::Execution("lifecycle callback panicked".into()))?;
264 let mut source = Box::pin(source);
265 let observed = tokio::select! {
266 biased;
267 _ = tokio::time::sleep_until(deadline) => {
268 Err(WorkflowFailure::Timeout("lifecycle observation expired".into()))
269 }
270 result = AssertUnwindSafe(poll_fn(|context| source.as_mut().poll(context))).catch_unwind() => {
271 match result {
272 Ok(result) => result.map_err(WorkflowFailure::hook),
273 Err(_) => Err(WorkflowFailure::Execution("lifecycle callback panicked".into())),
274 }
275 }
276 };
277 // The callback's future is still owned here when its observation ends.
278 // Preserve an earlier callback/timeout failure if destruction also fails.
279 let destroyed = catch_unwind(AssertUnwindSafe(|| drop(source)))
280 .map_err(|_| WorkflowFailure::Execution("lifecycle callback panicked".into()));
281 observed.and(destroyed)
282 }
283
284 /// Projects this observation into the pinned rurr tracking JSON records.
285 ///
286 /// This is an explicit content export: inputs, results, state and attributes
287 /// may be present. The caller owns transport and samples the observation time.
288 /// No record is retained and no hook is invoked. The complete encoded list
289 /// is bounded to 1 MiB. Timed records require UTC years 1970 through 9999.
290 ///
291 /// Execute boundaries, child completion and nonfirst partial items have no
292 /// pinned wire row and return an empty list. Step completion serializes its
293 /// temporary state projection through the process-global field encoders;
294 /// the authoritative checkpoint remains unchanged.
295 /// A step without an outcome has no pinned end-entry state and returns an
296 /// input error before invoking a field encoder.
297 ///
298 /// ```
299 /// use app_kit::workflow::{LifecycleEvent, RunId};
300 /// let run: RunId = "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?;
301 /// let event = LifecycleEvent::ExecuteStarted {
302 /// run: &run, occurrence: "00000000000040008000000000000001",
303 /// };
304 /// assert!(event.tracking_records(std::time::UNIX_EPOCH)?.is_empty());
305 /// # Ok::<(), app_kit::AppError>(())
306 /// ```
307 ///
308 /// # Errors
309 ///
310 /// Returns `InvalidInput` for a step with no outcome, missing graph source,
311 /// unsupported timestamp, malformed span attributes, or exceeded record bounds.
312 /// Codec errors propagate from state serialization. This explicit export may
313 /// contain caller content; it is separate from safe operational tracing.
314 pub fn tracking_records(&self, observed_at: SystemTime) -> Result<Vec<Value>, AppError> {
315 let records = match self {
316 Self::Registered {
317 workspace,
318 run,
319 occurrence: _,
320 graph,
321 parent,
322 lineage,
323 } => {
324 let logical_parent = lineage.map(|lineage| {
325 json!({
326 "type":"pointer_data", "app_id":lineage.source,
327 "sequence_id":lineage.action_ordinal, "partition_key":workspace,
328 })
329 });
330 let spawning_parent = parent.map(|parent| {
331 json!({
332 "type":"pointer_data", "app_id":parent.run,
333 "sequence_id":parent.action_ordinal, "partition_key":parent.workspace,
334 })
335 });
336 let mut records = vec![
337 graph.tracking_record()?,
338 json!({
339 "type":"application_metadata", "partition_key":workspace,
340 "parent_pointer":logical_parent, "spawning_parent_pointer":spawning_parent,
341 }),
342 ];
343 if lineage.is_some() || parent.is_some() {
344 let time = Self::tracking_time(observed_at)?;
345 for (kind, ordinal) in lineage
346 .map(|lineage| ("fork", lineage.action_ordinal))
347 .into_iter()
348 .chain(parent.map(|parent| ("spawn_start", Some(parent.action_ordinal))))
349 {
350 records.push(json!({
351 "type":"child_application_data", "event_type":kind,
352 "sequence_id":ordinal, "event_time":time,
353 "child":{"type":"pointer_data", "app_id":run,
354 "sequence_id":null, "partition_key":workspace},
355 }));
356 }
357 }
358 records
359 }
360 Self::StepStarted {
361 context, inputs, ..
362 } => vec![json!({
363 "type":"begin_entry", "start_time":Self::tracking_time(observed_at)?,
364 "action":context.node, "inputs":inputs, "sequence_id":context.action_ordinal,
365 })],
366 Self::StepFinished {
367 context,
368 outcome,
369 checkpoint,
370 } => {
371 let Some(outcome) = outcome else {
372 return Err(AppError::InvalidInput(
373 "step without an outcome has no tracking end-entry state".into(),
374 ));
375 };
376 // Pinned field encoders see the attempted sequence and committed
377 // predecessor, including on failure. They never mutate the
378 // authoritative checkpoint or the outer numeric wire ordinal.
379 let mut state = checkpoint.state.clone();
380 state
381 .values
382 .insert("__SEQUENCE_ID".into(), context.action_ordinal.into());
383 match &checkpoint.position {
384 Some(position) => {
385 state
386 .values
387 .insert("__PRIOR_STEP".into(), position.clone().into());
388 }
389 None => {
390 state.values.remove("__PRIOR_STEP");
391 }
392 }
393 let (result, exception) = match outcome {
394 Ok(outcome) => (Some(&outcome.result), None),
395 Err(failure) => (None, Some(failure.message())),
396 };
397 vec![json!({
398 "type":"end_entry", "end_time":Self::tracking_time(observed_at)?,
399 "action":context.node, "result":result, "exception":exception,
400 "state":state.serialize()?, "sequence_id":context.action_ordinal,
401 })]
402 }
403 Self::SpanStarted {
404 context,
405 id,
406 name,
407 parent,
408 dependencies,
409 } => vec![json!({
410 "type":"begin_span", "start_time":Self::tracking_time(observed_at)?,
411 "action_sequence_id":context.action_ordinal, "span_id":id,
412 "span_name":name, "parent_span_id":parent, "span_dependencies":dependencies,
413 })],
414 Self::SpanFinished { context, id } => vec![json!({
415 "type":"end_span", "end_time":Self::tracking_time(observed_at)?,
416 "action_sequence_id":context.action_ordinal, "span_id":id,
417 })],
418 Self::SpanAttribute {
419 context,
420 span,
421 key,
422 value,
423 tags,
424 } => {
425 if tags.values().any(|value| !value.is_string()) {
426 return Err(AppError::InvalidInput(
427 "tracking attribute tags must be strings".into(),
428 ));
429 }
430 vec![json!({
431 "type":"attribute", "key":key, "action_sequence_id":context.action_ordinal,
432 "span_id":span, "value":value, "tags":tags,
433 "time_logged":Self::tracking_time(observed_at)?,
434 })]
435 }
436 Self::StreamStarted { context, span } => vec![json!({
437 "type":"begin_stream", "action_sequence_id":context.action_ordinal,
438 "span_id":span, "stream_init_time":Self::tracking_time(observed_at)?,
439 })],
440 Self::StreamItem {
441 context,
442 span,
443 index: 0,
444 ..
445 } => vec![json!({
446 "type":"first_item_stream", "action_sequence_id":context.action_ordinal,
447 "span_id":span, "first_item_time":Self::tracking_time(observed_at)?,
448 })],
449 Self::StreamFinished {
450 context,
451 span,
452 items,
453 ..
454 } => vec![json!({
455 "type":"end_stream", "action_sequence_id":context.action_ordinal,
456 "span_id":span, "end_time":Self::tracking_time(observed_at)?,
457 "items_streamed":items,
458 })],
459 Self::ChildFinished { .. }
460 | Self::ExecuteStarted { .. }
461 | Self::ExecuteFinished { .. }
462 | Self::StreamItem { .. } => Vec::new(),
463 };
464 JsonBoundary::validate_size(&records)?;
465 Ok(records)
466 }
467
468 fn tracking_time(observed_at: SystemTime) -> Result<String, AppError> {
469 let elapsed = observed_at
470 .duration_since(UNIX_EPOCH)
471 .map_err(|_| AppError::InvalidInput("tracking timestamp precedes UNIX_EPOCH".into()))?;
472 if elapsed.as_secs() >= 253_402_300_800 {
473 return Err(AppError::InvalidInput(
474 "tracking timestamp exceeds year 9999".into(),
475 ));
476 }
477 // The checked interval bounds this Gregorian conversion to 8,030 years.
478 // There is no platform time API, new dependency or separate clock owner.
479 let mut days = elapsed.as_secs() / 86_400;
480 let mut year = 1970_u64;
481 let leap = loop {
482 let leap =
483 year.is_multiple_of(4) && (!year.is_multiple_of(100) || year.is_multiple_of(400));
484 let length = 365 + u64::from(leap);
485 if days < length {
486 break leap;
487 }
488 days -= length;
489 year += 1;
490 };
491 let months = [
492 31,
493 28 + u64::from(leap),
494 31,
495 30,
496 31,
497 30,
498 31,
499 31,
500 30,
501 31,
502 30,
503 31,
504 ];
505 let mut month = 1;
506 for length in months {
507 if days < length {
508 break;
509 }
510 days -= length;
511 month += 1;
512 }
513 let time = elapsed.as_secs() % 86_400;
514 Ok(format!(
515 "{year:04}-{month:02}-{:02}T{:02}:{:02}:{:02}.{:09}Z",
516 days + 1,
517 time / 3_600,
518 (time / 60) % 60,
519 time % 60,
520 elapsed.subsec_nanos()
521 ))
522 }
523}
524
525impl WorkflowFailure {
526 pub(crate) fn with_cause(self, cause: Option<Self>) -> Self {
527 match self {
528 Self::Hook {
529 message,
530 cause: prior,
531 } => Self::Hook {
532 message,
533 cause: prior.or_else(|| cause.map(Box::new)),
534 },
535 failure => failure,
536 }
537 }
538
539 fn hook(error: AppError) -> Self {
540 Self::Hook {
541 message: error.to_string(),
542 cause: None,
543 }
544 }
545
546 pub(crate) fn application_error(&self) -> AppError {
547 match self {
548 Self::Timeout(message) => AppError::Timeout(message.clone()),
549 _ => AppError::Integration(self.message().to_owned()),
550 }
551 }
552
553 pub(crate) fn message(&self) -> &str {
554 match self {
555 Self::Hook { message, .. } => message,
556 Self::Evaluation(message)
557 | Self::Action(message)
558 | Self::State(message)
559 | Self::Agent(message)
560 | Self::Child(message)
561 | Self::Execution(message)
562 | Self::Timeout(message)
563 | Self::Limit(message) => message,
564 }
565 }
566}