soma_ops/
event.rs

1use serde::{Deserialize, Serialize};
2use serde_json::Value;
3
4use crate::{
5    ActorRef, AuthorizationId, CorrelationId, EventId, OperationId, OperationName, OperationPlan,
6    OperationResult, ProducerRef, ProgressEvent, RedactionMetadata, TargetRef, Timestamp,
7    TraceContext, VerificationResult,
8};
9
10/// Canonical operation lifecycle event type.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
12#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
13#[serde(rename_all = "snake_case")]
14#[non_exhaustive]
15pub enum OperationEventType {
16    /// Intent was accepted before target resolution.
17    Requested,
18    /// A concrete plan was produced.
19    Planned,
20    /// Product policy authorized the exact operation and target.
21    Authorized,
22    /// External execution began and may now affect target state.
23    Started,
24    /// Bounded execution progress was observed.
25    Progressed,
26    /// Execution completed successfully, independent of verification.
27    Succeeded,
28    /// Execution failed.
29    Failed,
30    /// Execution was cancelled.
31    Cancelled,
32    /// Runtime state was independently verified or found inconclusive.
33    Verified,
34}
35
36/// Lifecycle-specific operation event payload.
37#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
38#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
39#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
40#[non_exhaustive]
41pub enum OperationEventPayload {
42    /// Request metadata safe to persist before target resolution.
43    Requested {
44        /// SHA-256 or equivalent digest of redacted request parameters.
45        parameters_digest: String,
46        /// Additional bounded, redacted request metadata.
47        #[serde(default)]
48        metadata: Value,
49    },
50    /// Immutable authorization-relevant plan.
51    Planned(OperationPlan),
52    /// Opaque reference to product-issued authorization evidence.
53    Authorized {
54        /// Authorization identity.
55        authorization_id: AuthorizationId,
56        /// Optional human-confirmation reference safe for audit.
57        #[serde(default, skip_serializing_if = "Option::is_none")]
58        confirmation_ref: Option<String>,
59    },
60    /// Concrete target at the point mutation may begin.
61    Started {
62        /// Bound topology revision when known.
63        #[serde(default, skip_serializing_if = "Option::is_none")]
64        topology_revision: Option<String>,
65    },
66    /// One bounded progress update.
67    Progressed(ProgressEvent),
68    /// Successful terminal execution result.
69    Succeeded(OperationResult),
70    /// Failed terminal execution result.
71    Failed(OperationResult),
72    /// Cancelled terminal execution result.
73    Cancelled(OperationResult),
74    /// Independent runtime verification.
75    Verified(VerificationResult),
76}
77
78impl OperationEventPayload {
79    /// Returns the event type corresponding to this payload.
80    #[must_use]
81    pub const fn event_type(&self) -> OperationEventType {
82        match self {
83            Self::Requested { .. } => OperationEventType::Requested,
84            Self::Planned(_) => OperationEventType::Planned,
85            Self::Authorized { .. } => OperationEventType::Authorized,
86            Self::Started { .. } => OperationEventType::Started,
87            Self::Progressed(_) => OperationEventType::Progressed,
88            Self::Succeeded(_) => OperationEventType::Succeeded,
89            Self::Failed(_) => OperationEventType::Failed,
90            Self::Cancelled(_) => OperationEventType::Cancelled,
91            Self::Verified(_) => OperationEventType::Verified,
92        }
93    }
94}
95
96/// Common envelope for operation lifecycle events.
97#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
98#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
99pub struct OperationEventEnvelope {
100    event_id: EventId,
101    event_version: u32,
102    event_type: OperationEventType,
103    occurred_at: Timestamp,
104    operation_id: OperationId,
105    operation: OperationName,
106    correlation_id: CorrelationId,
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    causation_id: Option<OperationId>,
109    #[serde(default, skip_serializing_if = "Option::is_none")]
110    actor: Option<ActorRef>,
111    #[serde(default, skip_serializing_if = "Option::is_none")]
112    target: Option<TargetRef>,
113    producer: ProducerRef,
114    #[serde(default)]
115    trace: TraceContext,
116    payload: OperationEventPayload,
117    #[serde(default)]
118    redaction: RedactionMetadata,
119}
120
121/// Backward-compatible short name for the canonical event envelope.
122pub type OperationEvent = OperationEventEnvelope;
123
124impl OperationEventEnvelope {
125    /// Creates a version-one event envelope and derives its event type from the payload.
126    #[must_use]
127    pub fn new(
128        occurred_at: Timestamp,
129        operation_id: OperationId,
130        operation: OperationName,
131        correlation_id: CorrelationId,
132        producer: ProducerRef,
133        payload: OperationEventPayload,
134    ) -> Self {
135        Self {
136            event_id: EventId::new(),
137            event_version: 1,
138            event_type: payload.event_type(),
139            occurred_at,
140            operation_id,
141            operation,
142            correlation_id,
143            causation_id: None,
144            actor: None,
145            target: None,
146            producer,
147            trace: TraceContext::default(),
148            payload,
149            redaction: RedactionMetadata::none(),
150        }
151    }
152
153    /// Records the operation that caused this event's operation.
154    #[must_use]
155    pub fn with_causation_id(mut self, causation_id: OperationId) -> Self {
156        self.causation_id = Some(causation_id);
157        self
158    }
159
160    /// Records the actor when known.
161    #[must_use]
162    pub fn with_actor(mut self, actor: ActorRef) -> Self {
163        self.actor = Some(actor);
164        self
165    }
166
167    /// Records the resolved target when known.
168    #[must_use]
169    pub fn with_target(mut self, target: TargetRef) -> Self {
170        self.target = Some(target);
171        self
172    }
173
174    /// Adds trace context.
175    #[must_use]
176    pub fn with_trace(mut self, trace: TraceContext) -> Self {
177        self.trace = trace;
178        self
179    }
180
181    /// Adds redaction metadata.
182    #[must_use]
183    pub fn with_redaction(mut self, redaction: RedactionMetadata) -> Self {
184        self.redaction = redaction;
185        self
186    }
187
188    /// Validates envelope, payload, target, and terminal-status consistency.
189    pub fn validate(&self) -> Result<(), EventError> {
190        validate_envelope(self)
191    }
192
193    /// Returns the stable event identity.
194    #[must_use]
195    pub fn event_id(&self) -> &EventId {
196        &self.event_id
197    }
198
199    /// Returns the schema version.
200    #[must_use]
201    pub const fn event_version(&self) -> u32 {
202        self.event_version
203    }
204
205    /// Returns the lifecycle event type.
206    #[must_use]
207    pub const fn event_type(&self) -> OperationEventType {
208        self.event_type
209    }
210
211    /// Returns event occurrence time.
212    #[must_use]
213    pub const fn occurred_at(&self) -> Timestamp {
214        self.occurred_at
215    }
216
217    /// Returns operation execution identity.
218    #[must_use]
219    pub fn operation_id(&self) -> &OperationId {
220        &self.operation_id
221    }
222
223    /// Returns canonical operation name.
224    #[must_use]
225    pub fn operation(&self) -> &OperationName {
226        &self.operation
227    }
228
229    /// Returns workflow correlation identity.
230    #[must_use]
231    pub fn correlation_id(&self) -> &CorrelationId {
232        &self.correlation_id
233    }
234
235    /// Returns target when resolved.
236    #[must_use]
237    pub fn target(&self) -> Option<&TargetRef> {
238        self.target.as_ref()
239    }
240
241    /// Returns producing component identity.
242    #[must_use]
243    pub fn producer(&self) -> &ProducerRef {
244        &self.producer
245    }
246
247    /// Returns lifecycle payload.
248    #[must_use]
249    pub fn payload(&self) -> &OperationEventPayload {
250        &self.payload
251    }
252
253    /// Returns redaction metadata.
254    #[must_use]
255    pub fn redaction(&self) -> &RedactionMetadata {
256        &self.redaction
257    }
258}
259
260fn require_target(target: Option<&TargetRef>) -> Result<(), EventError> {
261    target.map(|_| ()).ok_or(EventError::MissingTarget)
262}
263
264fn validate_terminal_payload(
265    envelope: &OperationEventEnvelope,
266    result: &OperationResult,
267    expected_status: crate::OperationStatus,
268) -> Result<(), EventError> {
269    require_target(envelope.target.as_ref())?;
270    if result.operation_id() != &envelope.operation_id
271        || result.operation() != &envelope.operation
272        || Some(result.target()) != envelope.target.as_ref()
273    {
274        return Err(EventError::PayloadIdentityMismatch);
275    }
276    if result.status() != expected_status {
277        return Err(EventError::TerminalStatusMismatch);
278    }
279    result
280        .validate()
281        .map_err(|_| EventError::InvalidTerminalResult)
282}
283
284fn validate_sha256(value: &str) -> Result<(), EventError> {
285    if value.len() == 64
286        && value
287            .bytes()
288            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
289    {
290        Ok(())
291    } else {
292        Err(EventError::InvalidParametersDigest)
293    }
294}
295
296/// Invalid lifecycle event envelope.
297#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
298#[non_exhaustive]
299pub enum EventError {
300    /// Event schema versions are one-based.
301    #[error("operation event version must be greater than zero")]
302    ZeroVersion,
303    /// The stored event type does not match its tagged payload.
304    #[error("operation event type does not match payload")]
305    EventTypeMismatch,
306    /// A lifecycle stage that requires a resolved target omitted it.
307    #[error("operation event requires a resolved target")]
308    MissingTarget,
309    /// Payload operation identity, name, or target differs from the envelope.
310    #[error("operation event payload identity does not match envelope")]
311    PayloadIdentityMismatch,
312    /// Terminal payload status differs from the event type.
313    #[error("terminal operation result status does not match event type")]
314    TerminalStatusMismatch,
315    /// The terminal result violates result invariants.
316    #[error("terminal operation result is invalid")]
317    InvalidTerminalResult,
318    /// The embedded plan fingerprint is invalid.
319    #[error("planned operation event contains an invalid plan fingerprint")]
320    InvalidPlanFingerprint,
321    /// Requested parameter digest is not lowercase SHA-256.
322    #[error("requested operation parameters digest must be lowercase SHA-256")]
323    InvalidParametersDigest,
324}
325
326fn validate_envelope(envelope: &OperationEventEnvelope) -> Result<(), EventError> {
327    if envelope.event_version == 0 {
328        return Err(EventError::ZeroVersion);
329    }
330    if envelope.event_type != envelope.payload.event_type() {
331        return Err(EventError::EventTypeMismatch);
332    }
333    match &envelope.payload {
334        OperationEventPayload::Requested {
335            parameters_digest, ..
336        } => validate_sha256(parameters_digest)?,
337        OperationEventPayload::Planned(plan) => {
338            require_target(envelope.target.as_ref())?;
339            if plan.operation_id() != &envelope.operation_id
340                || plan.operation() != &envelope.operation
341                || Some(plan.target()) != envelope.target.as_ref()
342            {
343                return Err(EventError::PayloadIdentityMismatch);
344            }
345            plan.validate_fingerprint()
346                .map_err(|_| EventError::InvalidPlanFingerprint)?;
347        }
348        OperationEventPayload::Authorized { .. }
349        | OperationEventPayload::Started { .. }
350        | OperationEventPayload::Verified(_) => {
351            require_target(envelope.target.as_ref())?;
352        }
353        OperationEventPayload::Progressed(progress) => {
354            require_target(envelope.target.as_ref())?;
355            if progress.operation_id() != &envelope.operation_id
356                || progress.operation() != &envelope.operation
357            {
358                return Err(EventError::PayloadIdentityMismatch);
359            }
360        }
361        OperationEventPayload::Succeeded(result) => {
362            validate_terminal_payload(envelope, result, crate::OperationStatus::Succeeded)?;
363        }
364        OperationEventPayload::Failed(result) => {
365            validate_terminal_payload(envelope, result, crate::OperationStatus::Failed)?;
366        }
367        OperationEventPayload::Cancelled(result) => {
368            validate_terminal_payload(envelope, result, crate::OperationStatus::Cancelled)?;
369        }
370    }
371    Ok(())
372}
373
374/// Sink for operation lifecycle events, implemented by embedded or remote adapters.
375pub trait EventSink: Send + Sync {
376    /// Sink-specific delivery error.
377    type Error;
378
379    /// Delivers one idempotent event envelope.
380    fn emit(&self, event: &OperationEventEnvelope) -> Result<(), Self::Error>;
381}
382
383#[cfg(test)]
384#[path = "event_tests.rs"]
385mod tests;
soma_ops/event.rs

soma_ops/
event.rs
