artifactize_core/runs/
watch.rs

1//! Shared change-driven Run stream. Notifications are hints; saved state is authority.
2use super::{Context, Finding, RunObservation, RunResultPage};
3use crate::types::{RequestId, RequestStatus, RunId};
4use serde::Serialize;
5use std::{collections::BTreeMap, future::Future, time::Duration};
6use tokio_util::sync::CancellationToken;
7
8#[derive(Debug, Serialize)]
9#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
10pub enum RunEvent {
11    /// A live observation is never a final validation verdict.
12    Update(RunObservation),
13    /// A saved eval result, including reuse. Provisional GREEN is not final approval.
14    EvalResult(Finding),
15    /// The Run's final validation; findings may continue through next_cursor.
16    Outcome(RunResultPage),
17}
18
19/// Replay a baseline, then consume shared state invalidations, not caller polling.
20/// Reattachment replays saved completed evals. Within this attachment, each saved
21/// result status is emitted once. Cancellation/deadline drops only the observer.
22pub async fn watch<F, R>(
23    context: &Context,
24    id: &RunId,
25    bound: Option<Duration>,
26    cancellation: CancellationToken,
27    mut emit: F,
28) -> Result<RunObservation, String>
29where
30    F: FnMut(RunEvent) -> R,
31    R: Future<Output = Result<(), String>>,
32{
33    let mut subscription = None;
34    let mut last = None;
35    let observing = async {
36        let changes = subscription.insert(crate::changes::Subscription::new(&context.state).await);
37        let mut seen: BTreeMap<RequestId, RequestStatus> = BTreeMap::new();
38        let mut previous = None;
39        loop {
40            let view = super::view(context, id).await?;
41            let current = super::status(context, id).await?;
42            // Ignore elapsed time: reconciliation must not invent progress.
43            let signature = (current.state, current.counts.clone());
44            if previous.as_ref() != Some(&signature) {
45                emit(RunEvent::Update(current.clone())).await?;
46                previous = Some(signature);
47            }
48            for request in &view.requests {
49                if (request.result().is_some() || request.error().is_some())
50                    && seen.get(&request.id) != Some(&request.status())
51                {
52                    emit(RunEvent::EvalResult(super::result::finding(request))).await?;
53                    seen.insert(request.id.clone(), request.status());
54                }
55            }
56            let ended = super::observe::state(&view.run, &view.requests).is_terminal();
57            last = Some(current);
58            if ended {
59                emit(RunEvent::Outcome(super::result(context, id, None).await?)).await?;
60                return Ok::<(), String>(());
61            }
62            changes.next_reconciled().await;
63        }
64    };
65    let result = {
66        tokio::pin!(observing);
67        tokio::select! {
68            result = &mut observing => result,
69            _ = cancellation.cancelled() => Err("Run observation was cancelled.".into()),
70            _ = async { match bound { Some(bound) => tokio::time::sleep(bound).await, None => std::future::pending().await } } => Ok(()),
71        }
72    };
73    let closed = match subscription {
74        Some(subscription) => subscription.close().await,
75        None => Ok(()),
76    };
77    result?;
78    closed?;
79    last.ok_or_else(|| "State unavailable within the observation bound.".into())
80}
artifactize_core/runs/watch.rs

artifactize_core/runs/
watch.rs
