starweaver_session/
records.rs

1//! Durable session and run records.
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
5use serde_json::Value;
6use sha2::{Digest, Sha256};
7use starweaver_context::ResumableState;
8use starweaver_core::{
9    CheckpointId, ConversationId, Metadata, RunId, RunLifecycle, SessionId, TaskId, TraceContext,
10    VersionedRecordError,
11};
12use starweaver_stream::{ReplayCursor, ReplayCursorFamily, ReplayScope};
13
14use crate::{input::InputPart, management::SessionDeletionFence};
15
16/// Current durable session status.
17#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
18#[serde(rename_all = "snake_case")]
19pub enum SessionStatus {
20    /// Session can accept work.
21    #[default]
22    Active,
23    /// Session is archived.
24    Archived,
25    /// Session reached a failed terminal state.
26    Failed,
27    /// Session is tombstoned. Retained evidence is not model-visible.
28    Deleted,
29}
30
31/// Durable run status composed from admission state and the shared runtime lifecycle.
32#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
33pub struct DurableRunStatus(Option<RunLifecycle>);
34
35/// Backward-compatible public name for the durable run status.
36pub type RunStatus = DurableRunStatus;
37
38#[allow(non_upper_case_globals)]
39impl DurableRunStatus {
40    /// Run is accepted and awaiting runtime admission.
41    pub const Queued: Self = Self(None);
42    /// Runtime initialization is in progress.
43    pub const Starting: Self = Self(Some(RunLifecycle::Starting));
44    /// Runtime is actively executing.
45    pub const Running: Self = Self(Some(RunLifecycle::Running));
46    /// Runtime is waiting for external work.
47    pub const Waiting: Self = Self(Some(RunLifecycle::Waiting));
48    /// Runtime completed successfully.
49    pub const Completed: Self = Self(Some(RunLifecycle::Completed));
50    /// Runtime failed.
51    pub const Failed: Self = Self(Some(RunLifecycle::Failed));
52    /// Runtime was cancelled or interrupted.
53    pub const Cancelled: Self = Self(Some(RunLifecycle::Cancelled));
54
55    /// Return the admitted runtime lifecycle, or `None` while queued.
56    #[must_use]
57    pub const fn lifecycle(self) -> Option<RunLifecycle> {
58        self.0
59    }
60
61    /// Return the stable flat wire name.
62    #[must_use]
63    pub const fn as_str(self) -> &'static str {
64        match self.0 {
65            None => "queued",
66            Some(lifecycle) => lifecycle.as_str(),
67        }
68    }
69
70    /// Return whether the run owns the active session slot.
71    #[must_use]
72    pub const fn is_active(self) -> bool {
73        matches!(
74            self.0,
75            None | Some(RunLifecycle::Starting | RunLifecycle::Running | RunLifecycle::Waiting)
76        )
77    }
78
79    /// Return whether the run reached a terminal lifecycle.
80    #[must_use]
81    pub const fn is_terminal(self) -> bool {
82        match self.0 {
83            Some(lifecycle) => lifecycle.is_terminal(),
84            None => false,
85        }
86    }
87}
88
89impl From<RunLifecycle> for DurableRunStatus {
90    fn from(value: RunLifecycle) -> Self {
91        Self(Some(value))
92    }
93}
94
95impl TryFrom<DurableRunStatus> for RunLifecycle {
96    type Error = QueuedRunStatus;
97
98    fn try_from(value: DurableRunStatus) -> Result<Self, Self::Error> {
99        value.0.ok_or(QueuedRunStatus)
100    }
101}
102
103impl Serialize for DurableRunStatus {
104    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
105    where
106        S: Serializer,
107    {
108        serializer.serialize_str(self.as_str())
109    }
110}
111
112impl<'de> Deserialize<'de> for DurableRunStatus {
113    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
114    where
115        D: Deserializer<'de>,
116    {
117        match String::deserialize(deserializer)?.as_str() {
118            "queued" => Ok(Self::Queued),
119            "starting" => Ok(Self::Starting),
120            "running" => Ok(Self::Running),
121            "waiting" => Ok(Self::Waiting),
122            "completed" => Ok(Self::Completed),
123            "failed" => Ok(Self::Failed),
124            "cancelled" => Ok(Self::Cancelled),
125            other => Err(D::Error::custom(format!("unknown run status: {other}"))),
126        }
127    }
128}
129
130/// Product-neutral diagnostic retained with a terminal durable run.
131#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
132pub struct RunTerminalError {
133    /// Stable `snake_case` category chosen by the producing boundary.
134    pub code: String,
135    /// Sanitized diagnostic suitable for durable retrieval.
136    pub message: String,
137}
138
139impl RunTerminalError {
140    /// Build a terminal diagnostic from a stable code and safe message.
141    #[must_use]
142    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
143        Self {
144            code: code.into(),
145            message: message.into(),
146        }
147    }
148}
149
150/// Atomic terminal status, output, and diagnostic projection.
151#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
152pub struct RunTerminalProjection {
153    /// Terminal durable status.
154    pub status: RunStatus,
155    /// User-visible final output preview, not an error transport.
156    #[serde(default, skip_serializing_if = "Option::is_none")]
157    pub output_preview: Option<String>,
158    /// Safe terminal diagnostic.
159    #[serde(default, skip_serializing_if = "Option::is_none")]
160    pub error: Option<RunTerminalError>,
161}
162
163impl RunTerminalProjection {
164    /// Build and validate a terminal projection.
165    ///
166    /// # Errors
167    ///
168    /// Returns an error for a non-terminal status, a failed status without a diagnostic,
169    /// a completed status with a diagnostic, or an empty diagnostic field.
170    pub fn try_new(
171        status: RunStatus,
172        output_preview: Option<String>,
173        error: Option<RunTerminalError>,
174    ) -> Result<Self, RunTerminalProjectionError> {
175        let projection = Self {
176            status,
177            output_preview,
178            error,
179        };
180        projection.validate()?;
181        Ok(projection)
182    }
183
184    /// Build a successful terminal projection.
185    #[must_use]
186    pub const fn completed(output_preview: Option<String>) -> Self {
187        Self {
188            status: RunStatus::Completed,
189            output_preview,
190            error: None,
191        }
192    }
193
194    /// Build a failed terminal projection with no output preview.
195    #[must_use]
196    pub const fn failed(error: RunTerminalError) -> Self {
197        Self {
198            status: RunStatus::Failed,
199            output_preview: None,
200            error: Some(error),
201        }
202    }
203
204    /// Build a cancelled terminal projection with an optional safe reason.
205    #[must_use]
206    pub const fn cancelled(error: Option<RunTerminalError>) -> Self {
207        Self {
208            status: RunStatus::Cancelled,
209            output_preview: None,
210            error,
211        }
212    }
213
214    /// Validate the projection as a new terminal write.
215    ///
216    /// Historical records may lack a diagnostic. Stores validate this invariant only when
217    /// terminalizing a non-terminal record, while preserving already-committed legacy evidence.
218    ///
219    /// # Errors
220    ///
221    /// Returns an error when the projection violates terminal status or diagnostic invariants.
222    pub fn validate(&self) -> Result<(), RunTerminalProjectionError> {
223        if !self.status.is_terminal() {
224            return Err(RunTerminalProjectionError::NonTerminalStatus(self.status));
225        }
226        if self.status == RunStatus::Failed && self.error.is_none() {
227            return Err(RunTerminalProjectionError::MissingFailureDiagnostic);
228        }
229        if self.status == RunStatus::Completed && self.error.is_some() {
230            return Err(RunTerminalProjectionError::UnexpectedSuccessDiagnostic);
231        }
232        if let Some(error) = self.error.as_ref() {
233            if error.code.is_empty() {
234                return Err(RunTerminalProjectionError::EmptyDiagnosticCode);
235            }
236            if error.message.is_empty() {
237                return Err(RunTerminalProjectionError::EmptyDiagnosticMessage);
238            }
239        }
240        Ok(())
241    }
242
243    /// Return whether this projection exactly matches a durable run record.
244    #[must_use]
245    pub fn matches(&self, run: &RunRecord) -> bool {
246        (run.status, &run.output_preview, &run.terminal_error)
247            == (self.status, &self.output_preview, &self.error)
248    }
249
250    /// Apply this complete projection to a run record.
251    pub fn apply_to(&self, run: &mut RunRecord) {
252        run.status = self.status;
253        run.output_preview.clone_from(&self.output_preview);
254        run.terminal_error.clone_from(&self.error);
255    }
256}
257
258/// Invalid new terminal run projection.
259#[derive(Clone, Debug, Eq, PartialEq)]
260pub enum RunTerminalProjectionError {
261    /// The supplied status is not terminal.
262    NonTerminalStatus(RunStatus),
263    /// A failed run omitted its diagnostic.
264    MissingFailureDiagnostic,
265    /// A completed run carried an error diagnostic.
266    UnexpectedSuccessDiagnostic,
267    /// A non-terminal run carried a stale terminal diagnostic.
268    UnexpectedNonTerminalDiagnostic(RunStatus),
269    /// The stable diagnostic category is empty.
270    EmptyDiagnosticCode,
271    /// The diagnostic message is empty.
272    EmptyDiagnosticMessage,
273}
274
275impl std::fmt::Display for RunTerminalProjectionError {
276    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
277        match self {
278            Self::NonTerminalStatus(status) => {
279                write!(formatter, "run status {} is not terminal", status.as_str())
280            }
281            Self::MissingFailureDiagnostic => {
282                formatter.write_str("failed run requires a terminal diagnostic")
283            }
284            Self::UnexpectedSuccessDiagnostic => {
285                formatter.write_str("completed run cannot carry a terminal diagnostic")
286            }
287            Self::UnexpectedNonTerminalDiagnostic(status) => write!(
288                formatter,
289                "non-terminal run status {} cannot carry a terminal diagnostic",
290                status.as_str()
291            ),
292            Self::EmptyDiagnosticCode => formatter.write_str("terminal diagnostic code is empty"),
293            Self::EmptyDiagnosticMessage => {
294                formatter.write_str("terminal diagnostic message is empty")
295            }
296        }
297    }
298}
299
300impl std::error::Error for RunTerminalProjectionError {}
301
302/// A queued durable run has not entered an executable runtime lifecycle.
303#[derive(Clone, Copy, Debug, Eq, PartialEq)]
304pub struct QueuedRunStatus;
305
306impl std::fmt::Display for QueuedRunStatus {
307    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
308        formatter.write_str("queued run has no runtime lifecycle")
309    }
310}
311
312impl std::error::Error for QueuedRunStatus {}
313
314/// Generic execution status for approval, deferred, checkpoint, and archive workflows.
315#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
316#[serde(rename_all = "snake_case")]
317pub enum ExecutionStatus {
318    /// Item has been created.
319    Pending,
320    /// Item is currently processing.
321    Running,
322    /// Item is waiting on an external decision or worker.
323    Waiting,
324    /// Item completed successfully.
325    Completed,
326    /// Item failed.
327    Failed,
328    /// Item was cancelled.
329    Cancelled,
330}
331
332/// Stable reference to an exported environment provider state.
333#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
334pub struct EnvironmentStateRef {
335    /// Environment provider name.
336    pub provider: String,
337    /// Stable provider state reference.
338    pub reference: String,
339    /// Provider state revision, hash, or generation.
340    #[serde(default, skip_serializing_if = "Option::is_none")]
341    pub revision: Option<String>,
342    /// Provider-specific metadata.
343    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
344    pub metadata: Metadata,
345}
346
347/// Stable reference to a persisted checkpoint.
348#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
349pub struct CheckpointRef {
350    /// Checkpoint id.
351    pub checkpoint_id: CheckpointId,
352    /// Run id.
353    pub run_id: RunId,
354    /// Checkpoint sequence within the run.
355    pub sequence: usize,
356    /// Runtime node name.
357    pub node: String,
358    /// Optional storage URI.
359    #[serde(default, skip_serializing_if = "Option::is_none")]
360    pub storage_ref: Option<String>,
361    /// Stream cursor captured with this checkpoint.
362    #[serde(default, skip_serializing_if = "Option::is_none")]
363    pub stream_cursor: Option<usize>,
364    /// Creation time.
365    pub created_at: DateTime<Utc>,
366    /// Checkpoint metadata.
367    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
368    pub metadata: Metadata,
369}
370
371/// Stable durable reference to a family-aware stream replay position.
372#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
373pub struct StreamCursorRef {
374    /// Canonical stream cursor. Family, scope, sequence, and backend position live here once.
375    pub position: ReplayCursor,
376    /// Creation time.
377    pub created_at: DateTime<Utc>,
378    /// Cursor metadata.
379    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
380    pub metadata: Metadata,
381}
382
383impl StreamCursorRef {
384    /// Build a durable reference from a canonical cursor.
385    #[must_use]
386    pub fn new(position: ReplayCursor) -> Self {
387        Self {
388            position,
389            created_at: Utc::now(),
390            metadata: Metadata::default(),
391        }
392    }
393
394    /// Return the cursor family.
395    #[must_use]
396    pub const fn family(&self) -> ReplayCursorFamily {
397        self.position.family
398    }
399
400    /// Return the cursor scope.
401    #[must_use]
402    pub const fn scope(&self) -> &ReplayScope {
403        &self.position.scope
404    }
405
406    /// Return the last observed sequence.
407    #[must_use]
408    pub const fn sequence(&self) -> usize {
409        self.position.sequence
410    }
411
412    /// Return whether two references address the same stream family and scope.
413    #[must_use]
414    pub fn same_stream(&self, other: &Self) -> bool {
415        self.family() == other.family() && self.scope() == other.scope()
416    }
417
418    /// Validate that this cursor belongs to the supplied run scope.
419    ///
420    /// # Errors
421    ///
422    /// Returns an error when the cursor addresses another run or a non-run scope.
423    pub fn validate_for_run(&self, run_id: &RunId) -> Result<(), StreamCursorRefError> {
424        let expected = ReplayScope::run(run_id.as_str());
425        if self.scope() != &expected {
426            return Err(StreamCursorRefError::WrongScope {
427                expected: expected.as_str().to_string(),
428                actual: self.scope().as_str().to_string(),
429            });
430        }
431        Ok(())
432    }
433
434    /// Validate that replacing an existing same-stream cursor does not regress.
435    ///
436    /// # Errors
437    ///
438    /// Returns an error when the proposed sequence is behind the current sequence.
439    pub fn validate_progression(&self, current: &Self) -> Result<(), StreamCursorRefError> {
440        if self.same_stream(current) && self.sequence() < current.sequence() {
441            return Err(StreamCursorRefError::SequenceRegression {
442                family: self.family(),
443                scope: self.scope().as_str().to_string(),
444                current: current.sequence(),
445                proposed: self.sequence(),
446            });
447        }
448        Ok(())
449    }
450}
451
452/// Invalid durable stream-cursor update.
453#[derive(Clone, Debug, Eq, PartialEq)]
454pub enum StreamCursorRefError {
455    /// Cursor scope does not identify the run being updated.
456    WrongScope {
457        /// Expected run scope.
458        expected: String,
459        /// Supplied scope.
460        actual: String,
461    },
462    /// Cursor sequence would move durable replay progress backwards.
463    SequenceRegression {
464        /// Cursor family.
465        family: ReplayCursorFamily,
466        /// Cursor scope.
467        scope: String,
468        /// Current sequence.
469        current: usize,
470        /// Proposed older sequence.
471        proposed: usize,
472    },
473}
474
475impl std::fmt::Display for StreamCursorRefError {
476    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
477        match self {
478            Self::WrongScope { expected, actual } => {
479                write!(
480                    formatter,
481                    "expected cursor scope {expected}, received {actual}"
482                )
483            }
484            Self::SequenceRegression {
485                family,
486                scope,
487                current,
488                proposed,
489            } => write!(
490                formatter,
491                "{} cursor for {scope} regressed from {current} to {proposed}",
492                family.as_str()
493            ),
494        }
495    }
496}
497
498impl std::error::Error for StreamCursorRefError {}
499
500#[derive(Deserialize)]
501#[serde(deny_unknown_fields)]
502struct CurrentStreamCursorRefWire {
503    position: ReplayCursor,
504    created_at: DateTime<Utc>,
505    #[serde(default)]
506    metadata: Metadata,
507}
508
509#[derive(Deserialize)]
510#[serde(deny_unknown_fields)]
511struct LegacyStreamCursorRefWire {
512    family: String,
513    scope: String,
514    sequence: usize,
515    #[serde(default)]
516    cursor: Option<String>,
517    created_at: DateTime<Utc>,
518    #[serde(default)]
519    metadata: Metadata,
520}
521
522#[derive(Deserialize)]
523#[serde(untagged)]
524enum StreamCursorRefWire {
525    Current(CurrentStreamCursorRefWire),
526    Legacy(LegacyStreamCursorRefWire),
527}
528
529impl<'de> Deserialize<'de> for StreamCursorRef {
530    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
531    where
532        D: Deserializer<'de>,
533    {
534        match StreamCursorRefWire::deserialize(deserializer)? {
535            StreamCursorRefWire::Current(current) => Ok(Self {
536                position: current.position,
537                created_at: current.created_at,
538                metadata: current.metadata,
539            }),
540            StreamCursorRefWire::Legacy(legacy) => {
541                let LegacyStreamCursorRefWire {
542                    family,
543                    scope,
544                    sequence,
545                    cursor,
546                    created_at,
547                    metadata,
548                } = legacy;
549                let family = match family.as_str() {
550                    "raw_runtime" => ReplayCursorFamily::RawRuntime,
551                    "display" => ReplayCursorFamily::Display,
552                    "replay_event" => ReplayCursorFamily::ReplayEvent,
553                    other => {
554                        return Err(D::Error::custom(format!(
555                            "unknown stream cursor family: {other}"
556                        )));
557                    }
558                };
559                let mut position =
560                    ReplayCursor::for_family(family, ReplayScope::from_string(scope), sequence);
561                position.backend_cursor = cursor;
562                Ok(Self {
563                    position,
564                    created_at,
565                    metadata,
566                })
567            }
568        }
569    }
570}
571
572/// Durable workspace provenance that does not confer live filesystem authority.
573#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
574#[serde(deny_unknown_fields)]
575pub struct WorkspaceProvenanceRef {
576    /// Stable execution-domain-local workspace identity.
577    pub workspace_id: String,
578    /// Safe user-facing label. This may preserve a legacy path as historical evidence only.
579    #[serde(default, skip_serializing_if = "Option::is_none")]
580    pub display_label: Option<String>,
581    /// Digest of the canonical provenance declaration used when this reference was recorded.
582    pub provenance_digest: String,
583}
584
585impl WorkspaceProvenanceRef {
586    /// Build a host-approved workspace provenance reference.
587    #[must_use]
588    pub fn new(
589        workspace_id: impl Into<String>,
590        display_label: Option<String>,
591        provenance_digest: impl Into<String>,
592    ) -> Self {
593        Self {
594            workspace_id: workspace_id.into(),
595            display_label,
596            provenance_digest: provenance_digest.into(),
597        }
598    }
599
600    /// Derive stable durable provenance for a canonical root in one execution domain.
601    ///
602    /// This records identity only. It never grants filesystem authority; a host must still match
603    /// it against an independently registered live workspace grant.
604    #[must_use]
605    pub fn for_execution_domain_root(
606        execution_domain_id: &str,
607        canonical_root: &str,
608        display_label: Option<String>,
609    ) -> Self {
610        let identity = workspace_identity_digest(
611            b"starweaver.workspace.id.v1\0",
612            execution_domain_id,
613            canonical_root,
614        );
615        let provenance = workspace_identity_digest(
616            b"starweaver.workspace.provenance.v1\0",
617            execution_domain_id,
618            canonical_root,
619        );
620        Self {
621            workspace_id: format!("workspace_{}", &identity[..32]),
622            display_label,
623            provenance_digest: format!("sha256:{provenance}"),
624        }
625    }
626
627    /// Migrate one legacy workspace string into unbound historical provenance.
628    ///
629    /// The derived `legacy:` identity is deliberately outside the host workspace-id namespace and
630    /// must never be resolved as a live grant. It only keeps migration and filtering deterministic.
631    #[must_use]
632    pub fn from_legacy_display(display: impl Into<String>) -> Self {
633        let display = display.into();
634        let mut digest = Sha256::new();
635        digest.update(b"starweaver.session.workspace_provenance.legacy.v1\0");
636        digest.update(display.as_bytes());
637        let digest = format!("sha256:{:x}", digest.finalize());
638        Self {
639            workspace_id: format!("legacy:{digest}"),
640            display_label: Some(display),
641            provenance_digest: digest,
642        }
643    }
644
645    /// Return whether this reference was synthesized from legacy display-only evidence.
646    #[must_use]
647    pub fn is_legacy_unbound(&self) -> bool {
648        self.workspace_id.starts_with("legacy:")
649    }
650
651    /// Return the safe value used by legacy list/filter projections.
652    #[must_use]
653    pub fn display_value(&self) -> &str {
654        self.display_label.as_deref().unwrap_or(&self.workspace_id)
655    }
656
657    /// Return whether a legacy workspace filter matches the stable id or display projection.
658    #[must_use]
659    pub fn matches_filter(&self, workspace: &str) -> bool {
660        self.workspace_id == workspace || self.display_label.as_deref() == Some(workspace)
661    }
662
663    /// Validate durable provenance identity without conferring live authority.
664    ///
665    /// # Errors
666    ///
667    /// Returns an error when required identity evidence is empty or a legacy identity is
668    /// internally inconsistent.
669    pub fn validate(&self) -> Result<(), &'static str> {
670        if self.workspace_id.trim().is_empty() || self.provenance_digest.trim().is_empty() {
671            return Err("workspace provenance identity is empty");
672        }
673        if self.is_legacy_unbound()
674            && (self.workspace_id != format!("legacy:{}", self.provenance_digest)
675                || self.display_label.as_deref().is_none_or(str::is_empty))
676        {
677            return Err("legacy workspace provenance is inconsistent");
678        }
679        Ok(())
680    }
681}
682
683fn workspace_identity_digest(
684    domain: &[u8],
685    execution_domain_id: &str,
686    canonical_root: &str,
687) -> String {
688    let mut digest = Sha256::new();
689    digest.update(domain);
690    digest.update(
691        u64::try_from(execution_domain_id.len())
692            .unwrap_or(u64::MAX)
693            .to_be_bytes(),
694    );
695    digest.update(execution_domain_id.as_bytes());
696    digest.update(
697        u64::try_from(canonical_root.len())
698            .unwrap_or(u64::MAX)
699            .to_be_bytes(),
700    );
701    digest.update(canonical_root.as_bytes());
702    format!("{:x}", digest.finalize())
703}
704
705/// Immutable runtime-configuration snapshot identity pinned at run admission.
706#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
707#[serde(deny_unknown_fields)]
708pub struct RuntimeConfigSnapshotRef {
709    /// Monotonic runtime-config generation.
710    pub generation: u64,
711    /// Opaque revision etag identifying the retained immutable declaration.
712    pub etag: String,
713    /// Digest of the fully resolved non-secret run materialization.
714    pub materialization_digest: String,
715}
716
717impl RuntimeConfigSnapshotRef {
718    /// Build a runtime-config snapshot reference.
719    #[must_use]
720    pub fn new(
721        generation: u64,
722        etag: impl Into<String>,
723        materialization_digest: impl Into<String>,
724    ) -> Self {
725        Self {
726            generation,
727            etag: etag.into(),
728            materialization_digest: materialization_digest.into(),
729        }
730    }
731
732    /// Validate an immutable runtime-config snapshot identity.
733    ///
734    /// # Errors
735    ///
736    /// Returns an error when the generation or required identity evidence is empty.
737    pub fn validate(&self) -> Result<(), &'static str> {
738        if self.generation == 0 {
739            return Err("runtime config generation must be positive");
740        }
741        if self.etag.trim().is_empty() || self.materialization_digest.trim().is_empty() {
742            return Err("runtime config snapshot identity is empty");
743        }
744        Ok(())
745    }
746}
747
748#[derive(Deserialize)]
749struct SessionRecordV1 {
750    #[serde(default)]
751    workspace: Option<String>,
752    #[serde(flatten)]
753    fields: serde_json::Map<String, Value>,
754}
755
756fn decode_session_record_v1(payload: Value) -> Result<SessionRecord, VersionedRecordError> {
757    let legacy =
758        serde_json::from_value::<SessionRecordV1>(payload).map_err(VersionedRecordError::Json)?;
759    let mut current = serde_json::from_value::<SessionRecord>(Value::Object(legacy.fields))
760        .map_err(VersionedRecordError::Json)?;
761    current.workspace = legacy
762        .workspace
763        .map(WorkspaceProvenanceRef::from_legacy_display);
764    Ok(current)
765}
766
767/// Durable session record.
768#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
769pub struct SessionRecord {
770    /// Session id.
771    pub session_id: SessionId,
772    /// Host-derived store/tenant namespace. Legacy records default to `local`.
773    #[serde(default = "default_session_namespace")]
774    pub namespace_id: String,
775    /// Host-derived owner/principal. Lineage and metadata never assign authority.
776    #[serde(default, skip_serializing_if = "Option::is_none")]
777    pub owner_id: Option<String>,
778    /// Monotonic optimistic concurrency revision.
779    #[serde(default = "initial_session_revision")]
780    pub revision: u64,
781    /// Deletion/continuation fence.
782    #[serde(default)]
783    pub deletion_fence: SessionDeletionFence,
784    /// User-facing title.
785    #[serde(default, skip_serializing_if = "Option::is_none")]
786    pub title: Option<String>,
787    /// Workspace provenance. This evidence never recreates a host-local live grant.
788    #[serde(default, skip_serializing_if = "Option::is_none")]
789    pub workspace: Option<WorkspaceProvenanceRef>,
790    /// Runtime profile or model profile name.
791    #[serde(default, skip_serializing_if = "Option::is_none")]
792    pub profile: Option<String>,
793    /// Session status.
794    #[serde(default)]
795    pub status: SessionStatus,
796    /// Last exported context state.
797    #[serde(default)]
798    pub state: ResumableState,
799    /// Latest exported environment state reference.
800    #[serde(default, skip_serializing_if = "Option::is_none")]
801    pub environment_state: Option<EnvironmentStateRef>,
802    /// Latest stream cursor refs by family.
803    #[serde(default, skip_serializing_if = "Vec::is_empty")]
804    pub stream_cursors: Vec<StreamCursorRef>,
805    /// Session trace context.
806    #[serde(default, skip_serializing_if = "TraceContext::is_empty")]
807    pub trace_context: TraceContext,
808    /// Parent session id for forks or delegated flows.
809    #[serde(default, skip_serializing_if = "Option::is_none")]
810    pub parent_session_id: Option<SessionId>,
811    /// Latest run in the session sequence.
812    #[serde(default, skip_serializing_if = "Option::is_none")]
813    pub head_run_id: Option<RunId>,
814    /// Latest completed run usable as continuation source.
815    #[serde(default, skip_serializing_if = "Option::is_none")]
816    pub head_success_run_id: Option<RunId>,
817    /// Currently queued, running, or waiting run.
818    #[serde(default, skip_serializing_if = "Option::is_none")]
819    pub active_run_id: Option<RunId>,
820    /// Creation time.
821    pub created_at: DateTime<Utc>,
822    /// Last update time.
823    pub updated_at: DateTime<Utc>,
824    /// Metadata.
825    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
826    pub metadata: Metadata,
827}
828
829fn default_session_namespace() -> String {
830    crate::LOCAL_SESSION_NAMESPACE.to_string()
831}
832
833const fn initial_session_revision() -> u64 {
834    1
835}
836
837const fn initial_run_revision() -> u64 {
838    1
839}
840
841impl starweaver_core::VersionedRecord for SessionRecord {
842    const SCHEMA: &'static str = "starweaver.session.session_record";
843    const VERSION: u32 = 2;
844    const ALLOW_BARE_V0: bool = true;
845
846    fn decode_version(version: u32, payload: Value) -> Result<Self, VersionedRecordError> {
847        match version {
848            1 => decode_session_record_v1(payload),
849            Self::VERSION => serde_json::from_value(payload).map_err(VersionedRecordError::Json),
850            actual => Err(VersionedRecordError::UnsupportedVersion {
851                schema: Self::SCHEMA,
852                supported: Self::VERSION,
853                actual,
854            }),
855        }
856    }
857
858    fn decode_bare_v0(payload: Value) -> Result<Self, VersionedRecordError> {
859        decode_session_record_v1(payload)
860    }
861}
862
863impl SessionRecord {
864    /// Validate workspace provenance carried by this durable session.
865    ///
866    /// # Errors
867    ///
868    /// Returns an error when the optional provenance is malformed.
869    pub fn validate_provenance(&self) -> Result<(), &'static str> {
870        self.workspace
871            .as_ref()
872            .map_or(Ok(()), WorkspaceProvenanceRef::validate)
873    }
874
875    /// Build a session record with default state.
876    #[must_use]
877    pub fn new(session_id: SessionId) -> Self {
878        let now = Utc::now();
879        Self {
880            session_id,
881            namespace_id: default_session_namespace(),
882            owner_id: None,
883            revision: initial_session_revision(),
884            deletion_fence: SessionDeletionFence::Stable,
885            title: None,
886            workspace: None,
887            profile: None,
888            status: SessionStatus::Active,
889            state: ResumableState::default(),
890            environment_state: None,
891            stream_cursors: Vec::new(),
892            trace_context: TraceContext::default(),
893            parent_session_id: None,
894            head_run_id: None,
895            head_success_run_id: None,
896            active_run_id: None,
897            created_at: now,
898            updated_at: now,
899            metadata: Metadata::default(),
900        }
901    }
902}
903
904#[derive(Deserialize)]
905struct RunRecordV1 {
906    #[serde(flatten)]
907    fields: serde_json::Map<String, Value>,
908}
909
910fn decode_run_record_v1(payload: Value) -> Result<RunRecord, VersionedRecordError> {
911    let legacy =
912        serde_json::from_value::<RunRecordV1>(payload).map_err(VersionedRecordError::Json)?;
913    serde_json::from_value::<RunRecord>(Value::Object(legacy.fields))
914        .map_err(VersionedRecordError::Json)
915}
916
917/// Durable run record.
918#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
919pub struct RunRecord {
920    /// Session id.
921    pub session_id: SessionId,
922    /// Run id.
923    pub run_id: RunId,
924    /// Monotonic optimistic-concurrency revision.
925    #[serde(default = "initial_run_revision")]
926    pub revision: u64,
927    /// Conversation id.
928    pub conversation_id: ConversationId,
929    /// User, API, or service input parts.
930    #[serde(default, skip_serializing_if = "Vec::is_empty")]
931    pub input: Vec<InputPart>,
932    /// Durable run status.
933    #[serde(default)]
934    pub status: RunStatus,
935    /// Final output preview.
936    #[serde(default, skip_serializing_if = "Option::is_none")]
937    pub output_preview: Option<String>,
938    /// Safe diagnostic for a failed or cancelled terminal run.
939    #[serde(default, skip_serializing_if = "Option::is_none")]
940    pub terminal_error: Option<RunTerminalError>,
941    /// Final structured output preview or summary.
942    #[serde(default, skip_serializing_if = "Value::is_null")]
943    pub structured_output: Value,
944    /// Latest checkpoint reference.
945    #[serde(default, skip_serializing_if = "Option::is_none")]
946    pub latest_checkpoint: Option<CheckpointRef>,
947    /// Latest environment state reference for this run.
948    #[serde(default, skip_serializing_if = "Option::is_none")]
949    pub environment_state: Option<EnvironmentStateRef>,
950    /// Latest stream cursor refs by family.
951    #[serde(default, skip_serializing_if = "Vec::is_empty")]
952    pub stream_cursors: Vec<StreamCursorRef>,
953    /// Trace context.
954    #[serde(default, skip_serializing_if = "TraceContext::is_empty")]
955    pub trace_context: TraceContext,
956    /// Monotonic order inside the session.
957    #[serde(default)]
958    pub sequence_no: usize,
959    /// Run snapshot used as continuation source.
960    #[serde(default, skip_serializing_if = "Option::is_none")]
961    pub restore_from_run_id: Option<RunId>,
962    /// Parent run identifier when this run is delegated from another run.
963    #[serde(default, skip_serializing_if = "Option::is_none")]
964    pub parent_run_id: Option<RunId>,
965    /// Parent-scoped delegated task identifier when this run executes a lightweight task.
966    #[serde(default, skip_serializing_if = "Option::is_none")]
967    pub parent_task_id: Option<TaskId>,
968    /// Trigger source such as cli, service, schedule, or delegated.
969    #[serde(default, skip_serializing_if = "Option::is_none")]
970    pub trigger_type: Option<String>,
971    /// Profile resolved for this run.
972    #[serde(default, skip_serializing_if = "Option::is_none")]
973    pub profile: Option<String>,
974    /// Immutable runtime-config snapshot pinned when this run was admitted.
975    #[serde(default, skip_serializing_if = "Option::is_none")]
976    pub config_snapshot: Option<RuntimeConfigSnapshotRef>,
977    /// Creation time.
978    pub created_at: DateTime<Utc>,
979    /// Last update time.
980    pub updated_at: DateTime<Utc>,
981    /// Metadata.
982    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
983    pub metadata: Metadata,
984}
985
986impl starweaver_core::VersionedRecord for RunRecord {
987    const SCHEMA: &'static str = "starweaver.session.run_record";
988    const VERSION: u32 = 2;
989    const ALLOW_BARE_V0: bool = true;
990
991    fn decode_version(version: u32, payload: Value) -> Result<Self, VersionedRecordError> {
992        match version {
993            1 => decode_run_record_v1(payload),
994            Self::VERSION => serde_json::from_value(payload).map_err(VersionedRecordError::Json),
995            actual => Err(VersionedRecordError::UnsupportedVersion {
996                schema: Self::SCHEMA,
997                supported: Self::VERSION,
998                actual,
999            }),
1000        }
1001    }
1002
1003    fn decode_bare_v0(payload: Value) -> Result<Self, VersionedRecordError> {
1004        decode_run_record_v1(payload)
1005    }
1006}
1007
1008impl RunRecord {
1009    /// Validate runtime-config provenance carried by this durable run.
1010    ///
1011    /// # Errors
1012    ///
1013    /// Returns an error when the optional snapshot identity is malformed.
1014    pub fn validate_provenance(&self) -> Result<(), &'static str> {
1015        self.config_snapshot
1016            .as_ref()
1017            .map_or(Ok(()), RuntimeConfigSnapshotRef::validate)
1018    }
1019
1020    /// Return the complete terminal projection when this record is terminal.
1021    ///
1022    /// Historical failed records may legitimately return a projection with no diagnostic.
1023    #[must_use]
1024    pub fn terminal_projection(&self) -> Option<RunTerminalProjection> {
1025        self.status.is_terminal().then(|| RunTerminalProjection {
1026            status: self.status,
1027            output_preview: self.output_preview.clone(),
1028            error: self.terminal_error.clone(),
1029        })
1030    }
1031
1032    /// Validate a newly persisted run state.
1033    ///
1034    /// Historical records are accepted when read, but every new write must carry a complete
1035    /// terminal projection and every non-terminal write must omit stale terminal diagnostics.
1036    ///
1037    /// # Errors
1038    ///
1039    /// Returns an error when status, output, and terminal diagnostic are inconsistent.
1040    pub fn validate_new_write(&self) -> Result<(), RunTerminalProjectionError> {
1041        self.terminal_projection().map_or_else(
1042            || {
1043                if self.terminal_error.is_some() {
1044                    Err(RunTerminalProjectionError::UnexpectedNonTerminalDiagnostic(
1045                        self.status,
1046                    ))
1047                } else {
1048                    Ok(())
1049                }
1050            },
1051            |terminal| terminal.validate(),
1052        )
1053    }
1054
1055    /// Normalize caller-provided state before a new admission is persisted.
1056    ///
1057    /// Admission always creates a queued execution. Any stale terminal projection from a reused
1058    /// record is discarded rather than becoming active-run state or client-visible output.
1059    pub fn normalize_for_admission(&mut self) {
1060        self.status = RunStatus::Queued;
1061        self.output_preview = None;
1062        self.terminal_error = None;
1063    }
1064
1065    /// Apply the legacy low-level status update contract.
1066    ///
1067    /// The legacy API cannot distinguish safe diagnostics from arbitrary caller text, so failed
1068    /// and cancelled updates discard their preview and persist a fixed generic diagnostic.
1069    pub fn apply_legacy_status_update(
1070        &mut self,
1071        status: RunStatus,
1072        output_preview: Option<String>,
1073    ) {
1074        self.status = status;
1075        match status {
1076            RunStatus::Failed => {
1077                self.output_preview = None;
1078                self.terminal_error = Some(RunTerminalError::new(
1079                    "legacy_status_update_failed",
1080                    "run failed",
1081                ));
1082            }
1083            RunStatus::Cancelled => {
1084                self.output_preview = None;
1085                self.terminal_error = Some(RunTerminalError::new(
1086                    "legacy_status_update_cancelled",
1087                    "run cancelled",
1088                ));
1089            }
1090            _ => {
1091                self.output_preview = output_preview;
1092                self.terminal_error = None;
1093            }
1094        }
1095    }
1096
1097    /// Build a run record for a session.
1098    #[must_use]
1099    pub fn new(session_id: SessionId, run_id: RunId, conversation_id: ConversationId) -> Self {
1100        let now = Utc::now();
1101        Self {
1102            session_id,
1103            run_id,
1104            revision: initial_run_revision(),
1105            conversation_id,
1106            input: Vec::new(),
1107            status: RunStatus::Queued,
1108            output_preview: None,
1109            terminal_error: None,
1110            structured_output: Value::Null,
1111            latest_checkpoint: None,
1112            environment_state: None,
1113            stream_cursors: Vec::new(),
1114            trace_context: TraceContext::default(),
1115            sequence_no: 0,
1116            restore_from_run_id: None,
1117            parent_run_id: None,
1118            parent_task_id: None,
1119            trigger_type: None,
1120            profile: None,
1121            config_snapshot: None,
1122            created_at: now,
1123            updated_at: now,
1124            metadata: Metadata::default(),
1125        }
1126    }
1127}
starweaver_session/records.rs

starweaver_session/
records.rs
