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