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

starweaver_session/
host_events.rs
