aivcs-core 0.2.0

Core library for AIVCS domain logic and orchestration
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

//! Graph lifecycle adapter: bridges domain `Event` types to `RunLedger` persistence.

use std::sync::Arc;

use oxidized_state::{
    ContentDigest, RunEvent, RunId, RunLedger, RunMetadata, RunSummary, StorageResult,
};

use crate::domain::run::{Event, EventKind};

/// Extract the snake_case kind string from an `EventKind` via its serde tag.
fn event_kind_str(kind: &EventKind) -> String {
    serde_json::to_value(kind)
        .ok()
        .and_then(|v| v["type"].as_str().map(str::to_string))
        .unwrap_or_else(|| "unknown".to_string())
}

/// Adapter that records graph lifecycle [`Event`]s into a [`RunLedger`].
///
/// Usage:
/// 1. Call [`GraphRunRecorder::start`] to create a new run.
/// 2. Call [`GraphRunRecorder::record`] for each domain event.
/// 3. Call [`GraphRunRecorder::finish_ok`] or [`GraphRunRecorder::finish_err`] to finalize.
pub struct GraphRunRecorder {
    ledger: Arc<dyn RunLedger>,
    run_id: RunId,
}

impl GraphRunRecorder {
    /// Start a new run in the ledger, returning a recorder bound to that run.
    pub async fn start(
        ledger: Arc<dyn RunLedger>,
        spec_digest: &ContentDigest,
        metadata: RunMetadata,
    ) -> StorageResult<Self> {
        let run_id = ledger.create_run(spec_digest, metadata.clone()).await?;
        crate::obs::emit_run_started(run_id.to_string().as_str(), &metadata.agent_name);
        Ok(Self { ledger, run_id })
    }

    /// Record a single domain event into the ledger.
    pub async fn record(&self, event: &Event) -> StorageResult<()> {
        let kind_str = event_kind_str(&event.kind);
        let run_event = RunEvent {
            seq: event.seq,
            kind: kind_str.clone(),
            payload: event.payload.clone(),
            timestamp: event.timestamp,
        };
        self.ledger.append_event(&self.run_id, run_event).await?;
        crate::obs::emit_event_appended(&self.run_id.to_string(), &kind_str, event.seq);
        Ok(())
    }

    /// Finalize the run as completed.
    pub async fn finish_ok(self, summary: RunSummary) -> StorageResult<()> {
        let duration_ms = summary.duration_ms;
        let total_events = summary.total_events;
        self.ledger.complete_run(&self.run_id, summary).await?;
        crate::obs::emit_run_finished(&self.run_id.to_string(), duration_ms, total_events, true);
        Ok(())
    }

    /// Finalize the run as failed.
    pub async fn finish_err(self, summary: RunSummary) -> StorageResult<()> {
        let duration_ms = summary.duration_ms;
        let total_events = summary.total_events;
        self.ledger.fail_run(&self.run_id, summary).await?;
        crate::obs::emit_run_finished(&self.run_id.to_string(), duration_ms, total_events, false);
        Ok(())
    }

    /// Return a reference to the run ID.
    pub fn run_id(&self) -> &RunId {
        &self.run_id
    }
}