starweaver_session/
host_events.rs

1//! Product-neutral durable host-event evidence and replay queries.
2
3use std::collections::BTreeSet;
4
5use chrono::{DateTime, Utc};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use sha2::{Digest, Sha256};
9use starweaver_core::{RunId, SessionId};
10
11use crate::{RunRecord, RunStatus, SessionStoreError, SessionStoreResult};
12
13/// Maximum number of durable host events returned by one storage query.
14pub const MAX_HOST_EVENT_PAGE_SIZE: usize = 500;
15
16/// Largest backend position shared by memory and `SQLite` implementations.
17pub const MAX_HOST_EVENT_POSITION: u64 = 9_223_372_036_854_775_807;
18
19/// Stable product-neutral event classes persisted by the session store.
20///
21/// Transport-owned subscription closure is intentionally absent: it contains connection-local
22/// cursor and delivery-sequence material and is not durable host evidence.
23#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
24#[serde(rename_all = "snake_case")]
25pub enum DurableHostEventClass {
26    /// A durable session projection changed.
27    SessionChanged,
28    /// A durable run projection changed.
29    RunChanged,
30    /// Durable run output became available.
31    OutputAvailable,
32    /// A safe incremental assistant transcript changed.
33    TranscriptChanged,
34    /// An approval projection changed.
35    ApprovalChanged,
36    /// A deferred-tool projection changed.
37    DeferredChanged,
38    /// A clarification projection changed.
39    ClarificationChanged,
40    /// An environment attachment projection changed.
41    EnvironmentChanged,
42    /// A durable operator-facing diagnostic was recorded.
43    Diagnostic,
44}
45
46impl DurableHostEventClass {
47    /// Every durable host-event class in canonical order.
48    pub const ALL: [Self; 9] = [
49        Self::SessionChanged,
50        Self::RunChanged,
51        Self::OutputAvailable,
52        Self::TranscriptChanged,
53        Self::ApprovalChanged,
54        Self::DeferredChanged,
55        Self::ClarificationChanged,
56        Self::EnvironmentChanged,
57        Self::Diagnostic,
58    ];
59
60    /// Return the stable storage identifier for this event class.
61    #[must_use]
62    pub const fn as_str(self) -> &'static str {
63        match self {
64            Self::SessionChanged => "session_changed",
65            Self::RunChanged => "run_changed",
66            Self::OutputAvailable => "output_available",
67            Self::TranscriptChanged => "transcript_changed",
68            Self::ApprovalChanged => "approval_changed",
69            Self::DeferredChanged => "deferred_changed",
70            Self::ClarificationChanged => "clarification_changed",
71            Self::EnvironmentChanged => "environment_changed",
72            Self::Diagnostic => "diagnostic",
73        }
74    }
75}
76
77/// Resource scope owned by one durable host event.
78#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
79#[serde(tag = "kind", rename_all = "snake_case")]
80pub enum DurableHostEventScope {
81    /// Host-global evidence.
82    Global,
83    /// Evidence belonging to one session.
84    Session {
85        /// Durable session identity.
86        session_id: SessionId,
87    },
88    /// Evidence belonging to one run in one session.
89    Run {
90        /// Durable session identity.
91        session_id: SessionId,
92        /// Durable run identity.
93        run_id: RunId,
94    },
95}
96
97impl DurableHostEventScope {
98    /// Build a session scope.
99    #[must_use]
100    pub const fn session(session_id: SessionId) -> Self {
101        Self::Session { session_id }
102    }
103
104    /// Build a run scope.
105    #[must_use]
106    pub const fn run(session_id: SessionId, run_id: RunId) -> Self {
107        Self::Run { session_id, run_id }
108    }
109
110    /// Return the stable scope-kind identifier.
111    #[must_use]
112    pub const fn kind(&self) -> &'static str {
113        match self {
114            Self::Global => "global",
115            Self::Session { .. } => "session",
116            Self::Run { .. } => "run",
117        }
118    }
119
120    /// Return the session identity carried by this scope, when present.
121    #[must_use]
122    pub const fn session_id(&self) -> Option<&SessionId> {
123        match self {
124            Self::Global => None,
125            Self::Session { session_id } | Self::Run { session_id, .. } => Some(session_id),
126        }
127    }
128
129    /// Return the run identity carried by this scope, when present.
130    #[must_use]
131    pub const fn run_id(&self) -> Option<&RunId> {
132        match self {
133            Self::Run { run_id, .. } => Some(run_id),
134            Self::Global | Self::Session { .. } => None,
135        }
136    }
137
138    /// Return whether a record scope is visible inside this requested scope.
139    ///
140    /// A global view contains all resources, a session view contains that session and its runs,
141    /// and a run view contains only that exact run.
142    #[must_use]
143    pub fn contains(&self, record_scope: &Self) -> bool {
144        match (self, record_scope) {
145            (Self::Global, _) => true,
146            (
147                Self::Session {
148                    session_id: expected,
149                },
150                Self::Session { session_id } | Self::Run { session_id, .. },
151            ) => expected == session_id,
152            (
153                Self::Run {
154                    session_id: expected_session,
155                    run_id: expected_run,
156                },
157                Self::Run { session_id, run_id },
158            ) => expected_session == session_id && expected_run == run_id,
159            (Self::Session { .. } | Self::Run { .. }, Self::Global | Self::Session { .. }) => false,
160        }
161    }
162
163    fn identity_components(&self) -> Vec<&str> {
164        match self {
165            Self::Global => vec!["global"],
166            Self::Session { session_id } => vec!["session", session_id.as_str()],
167            Self::Run { session_id, run_id } => {
168                vec!["run", session_id.as_str(), run_id.as_str()]
169            }
170        }
171    }
172}
173
174/// Stable deterministic identity for one logical transition's event publication.
175#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
176#[serde(transparent)]
177pub struct EventPublicationKey(String);
178
179impl EventPublicationKey {
180    /// Derive an unambiguous publication key from a transition identity and event ordinal.
181    ///
182    /// The transition identity must itself be stable across retries, for example a durable
183    /// mutation receipt, admission identity, or sealed run-evidence identity.
184    ///
185    /// # Errors
186    ///
187    /// Returns an error when `transition_identity` is empty.
188    pub fn derive(
189        transition_identity: &str,
190        ordinal: u32,
191        scope: &DurableHostEventScope,
192        event_class: DurableHostEventClass,
193    ) -> SessionStoreResult<Self> {
194        if transition_identity.is_empty() {
195            return Err(SessionStoreError::Failed(
196                "host event transition identity cannot be empty".to_string(),
197            ));
198        }
199        let ordinal = ordinal.to_string();
200        let mut components = vec![transition_identity, event_class.as_str(), ordinal.as_str()];
201        components.extend(scope.identity_components());
202        Ok(Self(format!(
203            "host-event-publication-sha256:{:x}",
204            framed_digest(components)
205        )))
206    }
207
208    /// Return the publication-key string.
209    #[must_use]
210    pub fn as_str(&self) -> &str {
211        &self.0
212    }
213}
214
215/// One view-independent host event waiting for durable-log materialization.
216#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
217pub struct PendingHostEventPublication {
218    /// Deterministic logical-publication identity.
219    pub publication_key: EventPublicationKey,
220    /// Deterministic public event identity retained across every view and retry.
221    pub event_id: String,
222    /// Resource scope owned by this event.
223    pub scope: DurableHostEventScope,
224    /// Stable event class used for storage-level eligibility filtering.
225    pub event_class: DurableHostEventClass,
226    /// Product-neutral object projection. It contains event fields but no wire cursor, caller,
227    /// authority, feature-set, or view material.
228    pub projection: Value,
229    /// Time at which the authoritative transition occurred.
230    pub occurred_at: DateTime<Utc>,
231}
232
233impl starweaver_core::VersionedRecord for PendingHostEventPublication {
234    const SCHEMA: &'static str = "starweaver.session.pending_host_event_publication";
235}
236
237impl PendingHostEventPublication {
238    /// Build deterministic publication evidence for one logical transition.
239    ///
240    /// # Errors
241    ///
242    /// Returns an error for an empty transition identity or a non-object projection.
243    pub fn new(
244        transition_identity: &str,
245        ordinal: u32,
246        scope: DurableHostEventScope,
247        event_class: DurableHostEventClass,
248        projection: Value,
249        occurred_at: DateTime<Utc>,
250    ) -> SessionStoreResult<Self> {
251        if !projection.is_object() {
252            return Err(SessionStoreError::Failed(
253                "durable host event projection must be a JSON object".to_string(),
254            ));
255        }
256        let publication_key =
257            EventPublicationKey::derive(transition_identity, ordinal, &scope, event_class)?;
258        let event_id = derived_event_id(&publication_key);
259        Ok(Self {
260            publication_key,
261            event_id,
262            scope,
263            event_class,
264            projection,
265            occurred_at,
266        })
267    }
268
269    /// Validate persisted publication evidence.
270    ///
271    /// # Errors
272    ///
273    /// Returns an error when identity or projection invariants are invalid.
274    pub fn validate(&self) -> SessionStoreResult<()> {
275        if self.publication_key.as_str().is_empty() || self.event_id.is_empty() {
276            return Err(SessionStoreError::Failed(
277                "durable host event identity cannot be empty".to_string(),
278            ));
279        }
280        if self.event_id != derived_event_id(&self.publication_key) {
281            return Err(SessionStoreError::Conflict(format!(
282                "host event identity does not match publication {}",
283                self.publication_key.as_str()
284            )));
285        }
286        if !self.projection.is_object() {
287            return Err(SessionStoreError::Failed(
288                "durable host event projection must be a JSON object".to_string(),
289            ));
290        }
291        Ok(())
292    }
293}
294
295/// Product-neutral run summary used by durable `run_changed` projections.
296#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
297#[serde(rename_all = "camelCase")]
298pub struct RunChangedSummary {
299    /// Run creation time.
300    pub created_at: DateTime<Utc>,
301    /// Stable diagnostic category when the run terminated with a diagnostic.
302    #[serde(default, skip_serializing_if = "Option::is_none")]
303    pub diagnostic_ref: Option<String>,
304    /// User-visible output preview.
305    #[serde(default, skip_serializing_if = "Option::is_none")]
306    pub output_preview: Option<String>,
307    /// Decimal string revision used without a transport-specific integer wrapper.
308    pub revision: String,
309    /// Durable run identity.
310    pub run_id: RunId,
311    /// Durable session identity.
312    pub session_id: SessionId,
313    /// Durable run status.
314    pub status: RunStatus,
315    /// Last authoritative update time.
316    pub updated_at: DateTime<Utc>,
317}
318
319impl From<&RunRecord> for RunChangedSummary {
320    fn from(run: &RunRecord) -> Self {
321        Self {
322            created_at: run.created_at,
323            diagnostic_ref: run.terminal_error.as_ref().map(|error| error.code.clone()),
324            output_preview: run.output_preview.clone(),
325            revision: run.revision.to_string(),
326            run_id: run.run_id.clone(),
327            session_id: run.session_id.clone(),
328            status: run.status,
329            updated_at: run.updated_at,
330        }
331    }
332}
333
334/// Product-neutral durable `run_changed` event projection.
335#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
336#[serde(rename_all = "camelCase")]
337pub struct RunChangedProjection {
338    /// Stable event discriminator.
339    pub kind: String,
340    /// Complete authoritative run summary.
341    pub run: RunChangedSummary,
342}
343
344/// Product-neutral durable `output_available` event projection.
345#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
346#[serde(rename_all = "camelCase")]
347pub struct OutputAvailableProjection {
348    /// Stable event discriminator.
349    pub kind: String,
350    /// Stable output identity derived from the durable run revision.
351    pub output_ref: String,
352    /// User-visible output preview.
353    pub preview: String,
354    /// Durable run identity.
355    pub run_id: RunId,
356    /// Durable session identity.
357    pub session_id: SessionId,
358}
359
360/// Closed safe assistant-transcript update carried by a durable host event.
361#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
362#[serde(
363    tag = "kind",
364    rename_all = "snake_case",
365    rename_all_fields = "camelCase"
366)]
367pub enum TranscriptUpdateProjection {
368    /// One assistant message started.
369    MessageStarted {
370        /// Stable run-local message identity.
371        message_id: String,
372        /// Closed role projection. This is always `assistant`.
373        role: String,
374    },
375    /// Public assistant text was appended.
376    TextAppended {
377        /// Stable run-local message identity.
378        message_id: String,
379        /// Safe public text delta.
380        delta: String,
381    },
382    /// One assistant message finished.
383    MessageFinished {
384        /// Stable run-local message identity.
385        message_id: String,
386    },
387}
388
389/// Product-neutral durable `transcript_changed` event projection.
390#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
391#[serde(rename_all = "camelCase")]
392pub struct TranscriptChangedProjection {
393    /// Stable event discriminator.
394    pub kind: String,
395    /// Durable run identity.
396    pub run_id: RunId,
397    /// Durable session identity.
398    pub session_id: SessionId,
399    /// Decimal-string sequence in the canonical run transcript.
400    pub transcript_sequence: String,
401    /// Closed transcript update.
402    pub update: TranscriptUpdateProjection,
403}
404
405/// Build one safe incremental transcript publication.
406///
407/// # Errors
408///
409/// Returns an error when the message identity or public text exceeds the host contract bounds, or
410/// when projection serialization/publication validation fails.
411pub fn transcript_changed_publication(
412    transition_identity: &str,
413    ordinal: u32,
414    session_id: SessionId,
415    run_id: RunId,
416    transcript_sequence: u64,
417    update: TranscriptUpdateProjection,
418    occurred_at: DateTime<Utc>,
419) -> SessionStoreResult<PendingHostEventPublication> {
420    let (message_id, delta) = match &update {
421        TranscriptUpdateProjection::MessageStarted { message_id, role } => {
422            if role != "assistant" {
423                return Err(SessionStoreError::Failed(
424                    "transcript role must be assistant".to_string(),
425                ));
426            }
427            (message_id, None)
428        }
429        TranscriptUpdateProjection::TextAppended { message_id, delta } => (message_id, Some(delta)),
430        TranscriptUpdateProjection::MessageFinished { message_id } => (message_id, None),
431    };
432    if message_id.is_empty() || message_id.chars().count() > 256 {
433        return Err(SessionStoreError::Failed(
434            "transcript message identity is outside host contract bounds".to_string(),
435        ));
436    }
437    if delta.is_some_and(|delta| delta.is_empty() || delta.chars().count() > 16_384) {
438        return Err(SessionStoreError::Failed(
439            "transcript text delta is outside host contract bounds".to_string(),
440        ));
441    }
442    let projection = serde_json::to_value(TranscriptChangedProjection {
443        kind: "transcript_changed".to_string(),
444        run_id: run_id.clone(),
445        session_id: session_id.clone(),
446        transcript_sequence: transcript_sequence.to_string(),
447        update,
448    })
449    .map_err(|error| SessionStoreError::Failed(error.to_string()))?;
450    PendingHostEventPublication::new(
451        transition_identity,
452        ordinal,
453        DurableHostEventScope::run(session_id, run_id),
454        DurableHostEventClass::TranscriptChanged,
455        projection,
456        occurred_at,
457    )
458}
459
460/// Build a durable `run_changed` publication from authoritative storage-domain state.
461///
462/// # Errors
463///
464/// Returns an error when projection serialization or publication identity validation fails.
465pub fn run_changed_publication(
466    transition_identity: &str,
467    ordinal: u32,
468    run: &RunRecord,
469) -> SessionStoreResult<PendingHostEventPublication> {
470    let projection = serde_json::to_value(RunChangedProjection {
471        kind: "run_changed".to_string(),
472        run: RunChangedSummary::from(run),
473    })
474    .map_err(|error| SessionStoreError::Failed(error.to_string()))?;
475    PendingHostEventPublication::new(
476        transition_identity,
477        ordinal,
478        DurableHostEventScope::run(run.session_id.clone(), run.run_id.clone()),
479        DurableHostEventClass::RunChanged,
480        projection,
481        run.updated_at,
482    )
483}
484
485/// Build a durable `output_available` publication when a run has a user-visible preview.
486///
487/// # Errors
488///
489/// Returns an error when projection serialization or publication identity validation fails.
490pub fn output_available_publication(
491    transition_identity: &str,
492    ordinal: u32,
493    run: &RunRecord,
494) -> SessionStoreResult<Option<PendingHostEventPublication>> {
495    let Some(preview) = run.output_preview.clone() else {
496        return Ok(None);
497    };
498    let projection = serde_json::to_value(OutputAvailableProjection {
499        kind: "output_available".to_string(),
500        output_ref: format!(
501            "run-output:{}:{}:{}",
502            run.session_id.as_str(),
503            run.run_id.as_str(),
504            run.revision
505        ),
506        preview,
507        run_id: run.run_id.clone(),
508        session_id: run.session_id.clone(),
509    })
510    .map_err(|error| SessionStoreError::Failed(error.to_string()))?;
511    PendingHostEventPublication::new(
512        transition_identity,
513        ordinal,
514        DurableHostEventScope::run(run.session_id.clone(), run.run_id.clone()),
515        DurableHostEventClass::OutputAvailable,
516        projection,
517        run.updated_at,
518    )
519    .map(Some)
520}
521
522/// Add authoritative run/output events unless the caller already supplied that class and scope.
523///
524/// This lets transport boundaries supply richer compatible projections while ensuring a bare
525/// run-evidence commit never leaves its authoritative state without atomic host publication.
526///
527/// # Errors
528///
529/// Returns an error when an automatically projected publication cannot be built.
530pub fn append_authoritative_run_publications<'a>(
531    publications: &mut Vec<PendingHostEventPublication>,
532    transition_identity: &str,
533    runs: impl IntoIterator<Item = &'a RunRecord>,
534) -> SessionStoreResult<()> {
535    for (index, run) in runs.into_iter().enumerate() {
536        let scope = DurableHostEventScope::run(run.session_id.clone(), run.run_id.clone());
537        let ordinal = u32::try_from(index).map_err(|error| {
538            SessionStoreError::Failed(format!("too many authoritative run publications: {error}"))
539        })?;
540        if !publications.iter().any(|publication| {
541            publication.scope == scope
542                && publication.event_class == DurableHostEventClass::RunChanged
543        }) {
544            publications.push(run_changed_publication(transition_identity, ordinal, run)?);
545        }
546        if run.output_preview.is_some()
547            && !publications.iter().any(|publication| {
548                publication.scope == scope
549                    && publication.event_class == DurableHostEventClass::OutputAvailable
550            })
551            && let Some(publication) =
552                output_available_publication(transition_identity, ordinal, run)?
553        {
554            publications.push(publication);
555        }
556        if run.status.is_terminal() && run.output_preview.is_some() {
557            let run_changed_index = publications.iter().position(|publication| {
558                publication.scope == scope
559                    && publication.event_class == DurableHostEventClass::RunChanged
560            });
561            let output_index = publications.iter().position(|publication| {
562                publication.scope == scope
563                    && publication.event_class == DurableHostEventClass::OutputAvailable
564            });
565            if let (Some(run_changed_index), Some(output_index)) = (run_changed_index, output_index)
566                && output_index > run_changed_index
567            {
568                let output = publications.remove(output_index);
569                publications.insert(run_changed_index, output);
570            }
571        }
572    }
573    Ok(())
574}
575
576/// One materialized durable host-event record.
577#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
578pub struct DurableHostEventRecord {
579    /// Storage-domain monotonic position. This is never exposed directly on the public wire.
580    pub position: u64,
581    /// Deterministic logical-publication identity.
582    pub publication_key: EventPublicationKey,
583    /// Stable public event identity.
584    pub event_id: String,
585    /// Resource scope owned by this event.
586    pub scope: DurableHostEventScope,
587    /// Stable event class.
588    pub event_class: DurableHostEventClass,
589    /// Product-neutral object projection.
590    pub projection: Value,
591    /// Time at which the authoritative transition occurred.
592    pub occurred_at: DateTime<Utc>,
593}
594
595impl starweaver_core::VersionedRecord for DurableHostEventRecord {
596    const SCHEMA: &'static str = "starweaver.session.durable_host_event";
597}
598
599impl DurableHostEventRecord {
600    /// Materialize one pending publication at a storage-assigned position.
601    #[must_use]
602    pub fn from_pending(position: u64, pending: PendingHostEventPublication) -> Self {
603        Self {
604            position,
605            publication_key: pending.publication_key,
606            event_id: pending.event_id,
607            scope: pending.scope,
608            event_class: pending.event_class,
609            projection: pending.projection,
610            occurred_at: pending.occurred_at,
611        }
612    }
613
614    /// Return the pending form used for exact-retry comparison.
615    #[must_use]
616    pub fn pending_projection(&self) -> PendingHostEventPublication {
617        PendingHostEventPublication {
618            publication_key: self.publication_key.clone(),
619            event_id: self.event_id.clone(),
620            scope: self.scope.clone(),
621            event_class: self.event_class,
622            projection: self.projection.clone(),
623            occurred_at: self.occurred_at,
624        }
625    }
626}
627
628/// Bounded storage query over one resource scope and an admitted set of event classes.
629#[derive(Clone, Debug, Eq, PartialEq)]
630pub struct DurableHostEventQuery {
631    /// Exact requested scope. Containment is applied by [`DurableHostEventScope::contains`].
632    pub scope: DurableHostEventScope,
633    /// Eligible event classes. Empty sets are rejected.
634    pub event_classes: BTreeSet<DurableHostEventClass>,
635    /// Return eligible records strictly after this backend position.
636    pub after_position: Option<u64>,
637    /// Maximum records returned, from 1 through [`MAX_HOST_EVENT_PAGE_SIZE`].
638    pub limit: usize,
639}
640
641impl DurableHostEventQuery {
642    /// Build and validate a replay query.
643    ///
644    /// # Errors
645    ///
646    /// Returns an error for an empty class set or invalid page size.
647    pub fn new(
648        scope: DurableHostEventScope,
649        event_classes: impl IntoIterator<Item = DurableHostEventClass>,
650        after_position: Option<u64>,
651        limit: usize,
652    ) -> SessionStoreResult<Self> {
653        let event_classes = event_classes.into_iter().collect::<BTreeSet<_>>();
654        if event_classes.is_empty() {
655            return Err(SessionStoreError::Failed(
656                "durable host event query requires at least one event class".to_string(),
657            ));
658        }
659        if !(1..=MAX_HOST_EVENT_PAGE_SIZE).contains(&limit) {
660            return Err(SessionStoreError::Failed(format!(
661                "durable host event query limit must be between 1 and {MAX_HOST_EVENT_PAGE_SIZE}"
662            )));
663        }
664        if after_position.is_some_and(|position| position > MAX_HOST_EVENT_POSITION) {
665            return Err(SessionStoreError::Failed(format!(
666                "durable host event position exceeds {MAX_HOST_EVENT_POSITION}"
667            )));
668        }
669        Ok(Self {
670            scope,
671            event_classes,
672            after_position,
673            limit,
674        })
675    }
676}
677
678/// One bounded, eligibility-filtered durable host-event page.
679#[derive(Clone, Debug, Eq, PartialEq)]
680pub struct DurableHostEventPage {
681    /// Eligible records in durable order.
682    pub records: Vec<DurableHostEventRecord>,
683    /// Last eligible backend position, or the requested start position when the page is empty.
684    pub next_position: Option<u64>,
685    /// Whether another eligible record exists after `next_position`.
686    pub has_more: bool,
687}
688
689fn derived_event_id(publication_key: &EventPublicationKey) -> String {
690    format!(
691        "event-sha256:{:x}",
692        framed_digest([publication_key.as_str()])
693    )
694}
695
696fn framed_digest<'a>(components: impl IntoIterator<Item = &'a str>) -> impl std::fmt::LowerHex {
697    let mut digest = Sha256::new();
698    for component in components {
699        digest.update(component.len().to_string().as_bytes());
700        digest.update(b":");
701        digest.update(component.as_bytes());
702        digest.update(b";");
703    }
704    digest.finalize()
705}
706
707#[cfg(test)]
708#[allow(clippy::expect_used)]
709mod tests {
710    use chrono::Utc;
711    use serde_json::json;
712    use starweaver_core::{ConversationId, RunId, SessionId};
713
714    use crate::{RunRecord, RunStatus};
715
716    use super::{
717        DurableHostEventClass, DurableHostEventQuery, DurableHostEventScope,
718        PendingHostEventPublication, append_authoritative_run_publications,
719        output_available_publication, run_changed_publication,
720    };
721
722    #[test]
723    fn publication_identity_is_stable_and_framed() {
724        let occurred_at = Utc::now();
725        let first = PendingHostEventPublication::new(
726            "a:b",
727            0,
728            DurableHostEventScope::session(SessionId::from_string("c")),
729            DurableHostEventClass::SessionChanged,
730            json!({"revision": "1"}),
731            occurred_at,
732        )
733        .expect("first publication");
734        let retry = PendingHostEventPublication::new(
735            "a:b",
736            0,
737            DurableHostEventScope::session(SessionId::from_string("c")),
738            DurableHostEventClass::SessionChanged,
739            json!({"revision": "1"}),
740            occurred_at,
741        )
742        .expect("retry publication");
743        let ambiguous_without_framing = PendingHostEventPublication::new(
744            "a",
745            0,
746            DurableHostEventScope::session(SessionId::from_string("b:c")),
747            DurableHostEventClass::SessionChanged,
748            json!({"revision": "1"}),
749            occurred_at,
750        )
751        .expect("second publication");
752
753        assert_eq!(first, retry);
754        assert_ne!(
755            first.publication_key,
756            ambiguous_without_framing.publication_key
757        );
758        assert_ne!(first.event_id, ambiguous_without_framing.event_id);
759    }
760
761    #[test]
762    fn authoritative_run_projections_are_product_neutral_wire_shapes() {
763        let mut run = RunRecord::new(
764            SessionId::from_string("session-event"),
765            RunId::from_string("run-event"),
766            ConversationId::from_string("conversation-event"),
767        );
768        run.status = RunStatus::Completed;
769        run.output_preview = Some("ready".to_string());
770        run.revision = 7;
771
772        let changed =
773            run_changed_publication("run-transition", 0, &run).expect("run changed publication");
774        assert_eq!(changed.projection["kind"], json!("run_changed"));
775        assert_eq!(changed.projection["run"]["revision"], json!("7"));
776        assert_eq!(changed.projection["run"]["runId"], json!("run-event"));
777        assert_eq!(changed.projection["run"]["status"], json!("completed"));
778
779        let output = output_available_publication("run-transition", 0, &run)
780            .expect("output publication")
781            .expect("preview produces output event");
782        assert_eq!(output.projection["kind"], json!("output_available"));
783        assert_eq!(output.projection["preview"], json!("ready"));
784        assert_eq!(output.projection["runId"], json!("run-event"));
785        assert_eq!(output.projection["sessionId"], json!("session-event"));
786    }
787
788    #[test]
789    fn terminal_output_is_published_before_the_terminal_run_event() {
790        let mut run = RunRecord::new(
791            SessionId::from_string("session-terminal"),
792            RunId::from_string("run-terminal"),
793            ConversationId::from_string("conversation-terminal"),
794        );
795        run.status = RunStatus::Completed;
796        run.output_preview = Some("complete".to_string());
797
798        let preseeded =
799            run_changed_publication("terminal-transition", 0, &run).expect("preseeded run event");
800        for mut publications in [Vec::new(), vec![preseeded]] {
801            append_authoritative_run_publications(&mut publications, "terminal-transition", [&run])
802                .expect("authoritative terminal publications");
803            assert_eq!(
804                publications
805                    .iter()
806                    .map(|publication| publication.event_class)
807                    .collect::<Vec<_>>(),
808                vec![
809                    DurableHostEventClass::OutputAvailable,
810                    DurableHostEventClass::RunChanged,
811                ]
812            );
813        }
814    }
815
816    #[test]
817    fn scope_containment_is_hierarchical() {
818        let session_id = SessionId::from_string("session-1");
819        let run = DurableHostEventScope::run(session_id.clone(), RunId::from_string("run-1"));
820        assert!(DurableHostEventScope::Global.contains(&run));
821        assert!(DurableHostEventScope::session(session_id).contains(&run));
822        assert!(run.contains(&run));
823        assert!(!DurableHostEventScope::session(SessionId::from_string("other")).contains(&run));
824    }
825
826    #[test]
827    fn query_rejects_empty_classes_and_unbounded_limits() {
828        assert!(DurableHostEventQuery::new(DurableHostEventScope::Global, [], None, 10).is_err());
829        assert!(
830            DurableHostEventQuery::new(
831                DurableHostEventScope::Global,
832                [DurableHostEventClass::Diagnostic],
833                None,
834                501,
835            )
836            .is_err()
837        );
838        assert!(
839            DurableHostEventQuery::new(
840                DurableHostEventScope::Global,
841                [DurableHostEventClass::Diagnostic],
842                Some(super::MAX_HOST_EVENT_POSITION + 1),
843                1,
844            )
845            .is_err()
846        );
847    }
848
849    #[test]
850    fn publication_validation_rejects_forged_event_identity() {
851        let mut publication = PendingHostEventPublication::new(
852            "transition",
853            0,
854            DurableHostEventScope::Global,
855            DurableHostEventClass::Diagnostic,
856            json!({"code": "test"}),
857            Utc::now(),
858        )
859        .expect("publication");
860        publication.event_id = "caller-selected".to_string();
861        assert!(publication.validate().is_err());
862    }
863}
starweaver_session/host_events.rs

starweaver_session/
host_events.rs
