starweaver_session/
publication.rs

1//! Durable stream-publication outbox contracts.
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5use sha2::{Digest, Sha256};
6use starweaver_core::{RunId, SessionId};
7use starweaver_stream::{AgentStreamRecord, DisplayMessage, ReplayEvent, ReplaySnapshot};
8
9/// External sink families selected for reliable publication.
10#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
11pub struct StreamPublicationTargets {
12    /// Publish raw and projected records to a [`starweaver_stream::StreamArchive`].
13    pub archive: bool,
14    /// Publish typed events to a [`starweaver_stream::ReplayEventLog`].
15    pub replay: bool,
16}
17
18impl StreamPublicationTargets {
19    /// Build target flags from configured sinks.
20    #[must_use]
21    pub const fn new(archive: bool, replay: bool) -> Self {
22        Self { archive, replay }
23    }
24
25    /// Return true when no external sink was selected.
26    #[must_use]
27    pub const fn is_empty(self) -> bool {
28        !self.archive && !self.replay
29    }
30}
31
32/// One transactionally enqueued stream-publication batch.
33#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
34pub struct PendingStreamPublication {
35    /// Stable outbox identity. One sealed run evidence bundle owns one publication.
36    pub publication_id: String,
37    /// Session that owns the evidence.
38    pub session_id: SessionId,
39    /// Run that owns the evidence.
40    pub run_id: RunId,
41    /// Raw runtime records for the archive sink.
42    #[serde(default, skip_serializing_if = "Vec::is_empty")]
43    pub stream_records: Vec<AgentStreamRecord>,
44    /// Projected display messages for the archive sink.
45    #[serde(default, skip_serializing_if = "Vec::is_empty")]
46    pub display_messages: Vec<DisplayMessage>,
47    /// Typed replay events for the replay-log sink.
48    #[serde(default, skip_serializing_if = "Vec::is_empty")]
49    pub replay_events: Vec<ReplayEvent>,
50    /// Optional compact display snapshot for the archive sink.
51    #[serde(default, skip_serializing_if = "Option::is_none")]
52    pub display_snapshot: Option<ReplaySnapshot>,
53    /// Whether archive delivery still requires acknowledgement.
54    pub archive_pending: bool,
55    /// Whether replay-log delivery still requires acknowledgement.
56    pub replay_pending: bool,
57    /// Time at which the evidence transaction enqueued this batch.
58    pub created_at: DateTime<Utc>,
59}
60
61impl starweaver_core::VersionedRecord for PendingStreamPublication {
62    const SCHEMA: &'static str = "starweaver.session.stream_publication";
63}
64
65fn publication_id(session_id: &SessionId, run_id: &RunId) -> String {
66    let mut digest = Sha256::new();
67    for component in [session_id.as_str(), run_id.as_str()] {
68        digest.update(component.len().to_string().as_bytes());
69        digest.update(b":");
70        digest.update(component.as_bytes());
71        digest.update(b";");
72    }
73    format!("publication-sha256:{:x}", digest.finalize())
74}
75
76impl PendingStreamPublication {
77    /// Build a deterministic publication from one run's sealed evidence.
78    #[must_use]
79    pub fn new(
80        session_id: SessionId,
81        run_id: RunId,
82        targets: StreamPublicationTargets,
83        created_at: DateTime<Utc>,
84    ) -> Self {
85        let publication_id = publication_id(&session_id, &run_id);
86        Self {
87            publication_id,
88            session_id,
89            run_id,
90            stream_records: Vec::new(),
91            display_messages: Vec::new(),
92            replay_events: Vec::new(),
93            display_snapshot: None,
94            archive_pending: targets.archive,
95            replay_pending: targets.replay,
96            created_at,
97        }
98    }
99
100    /// Return true when every configured sink acknowledged the batch.
101    #[must_use]
102    pub const fn is_complete(&self) -> bool {
103        !self.archive_pending && !self.replay_pending
104    }
105}
106
107/// One independently acknowledged external stream sink.
108#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
109#[serde(rename_all = "snake_case")]
110pub enum StreamPublicationTarget {
111    /// Raw records, display messages, and snapshots in a stream archive.
112    Archive,
113    /// Typed events in a replay event log.
114    Replay,
115}
116
117#[cfg(test)]
118mod tests {
119    use chrono::Utc;
120    use starweaver_core::{RunId, SessionId};
121
122    use super::{PendingStreamPublication, StreamPublicationTargets};
123
124    #[test]
125    fn publication_identity_is_unambiguous_for_delimiter_bearing_ids() {
126        let first = PendingStreamPublication::new(
127            SessionId::from_string("a:b"),
128            RunId::from_string("c"),
129            StreamPublicationTargets::new(true, true),
130            Utc::now(),
131        );
132        let second = PendingStreamPublication::new(
133            SessionId::from_string("a"),
134            RunId::from_string("b:c"),
135            StreamPublicationTargets::new(true, true),
136            Utc::now(),
137        );
138        assert_ne!(first.publication_id, second.publication_id);
139    }
140}
starweaver_session/publication.rs

starweaver_session/
publication.rs
