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

artifactize_core/runs/
watch.rs
