aion_proto/

events.rs

//! Event-streaming wire types.

use crate::convert::{
    ProtoRunId, ProtoWorkflowId, ProtoWorkflowStatus, WireEnvelope, decode_core_value,
    encode_core_value,
};
use crate::error::WireError;

/// Proto representation of an event subscription request.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct SubscriptionRequest {
    /// Requested subscription model.
    #[prost(oneof = "subscription_request::Subscription", tags = "1, 2, 3, 4, 5")]
    pub subscription: Option<subscription_request::Subscription>,
}

/// Types nested under [`SubscriptionRequest`].
pub mod subscription_request {
    /// Proto oneof for the available subscription models.
    #[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Oneof)]
    pub enum Subscription {
        /// Events for a single workflow in the caller's namespace.
        #[prost(message, tag = "1")]
        PerWorkflow(super::PerWorkflowSubscription),
        /// Events matching optional selectors scoped by the caller's namespace.
        #[prost(message, tag = "2")]
        Filtered(super::FilteredSubscription),
        /// All events visible in the caller's namespace.
        #[prost(message, tag = "3")]
        Firehose(super::FirehoseSubscription),
        /// Cluster topology/ownership deltas (WS3). Deployment-scoped, not
        /// namespace-scoped: authorized by the caller's deploy grant, not a
        /// namespace grant. This is a NEW ARM of the existing single
        /// subscription frame — the socket remains one-subscription-per-socket;
        /// a client wanting both workflow and cluster streams opens two
        /// `/events/stream` sockets (the second socket is honest and trivially
        /// supported, unlike a non-existent multiplexing layer).
        #[prost(message, tag = "4")]
        Cluster(super::ClusterSubscription),
        /// Agent-observability transcript for one `(workflow, activity, attempt)`
        /// (NOI-5b). Namespace-scoped exactly like [`Self::PerWorkflow`] — the
        /// transcript belongs to the workflow the activity runs under and is
        /// authorized by the caller's namespace grant, never the deploy grant.
        /// Like the other arms it is a NEW ARM of the single subscription frame;
        /// a client wanting both a workflow stream and a transcript opens two
        /// `/events/stream` sockets.
        #[prost(message, tag = "5")]
        Transcript(super::TranscriptSubscription),
    }
}

/// Subscribe to the cluster topology/ownership delta stream (WS3).
///
/// Carries an `after_seq` resume cursor that suppresses the in-flight broadcast
/// backlog on reconnect (cluster history is non-durable; a long disconnect
/// surfaces `cluster_lagged` and the client re-requests a snapshot). Unlike the
/// workflow subscriptions this carries no namespace — cluster topology is
/// deployment-scoped.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ClusterSubscription {
    /// Highest `cluster_seq` the client has already applied; the server drops
    /// buffered deltas with `cluster_seq <= after_seq` so a reconnect does not
    /// re-deliver them. `0` (the default) requests the full in-flight backlog.
    #[prost(uint64, tag = "1")]
    pub after_seq: u64,
}

/// Server -> client frame wrapping a single [`aion_core::ClusterEvent`] on the
/// cluster channel.
///
/// Mirrors [`StreamedEvent`] for the cluster path: the inner `aion-core` type is
/// the only thing that crosses the ts-rs boundary; this outer envelope is
/// hand-decoded on the TS side (same contract as the workflow path).
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StreamedClusterEvent {
    /// Frame discriminator pinned to `"cluster_event"` so the ops console's
    /// hand-written frame parser can branch a cluster delta apart from a
    /// `cluster_snapshot` priming reply or an `{"error": ...}` terminal frame.
    pub kind: String,
    /// The cluster delta.
    pub event: aion_core::ClusterEvent,
}

impl StreamedClusterEvent {
    /// Frame discriminator value for a live cluster delta.
    pub const KIND: &'static str = "cluster_event";

    /// Wrap a cluster event in its server->client frame.
    #[must_use]
    pub fn new(event: aion_core::ClusterEvent) -> Self {
        Self {
            kind: Self::KIND.to_owned(),
            event,
        }
    }
}

/// Server -> client priming frame carrying the calm-state [`aion_core::ClusterSnapshot`].
///
/// Sent once at the head of a cluster subscription before any live delta so the
/// ops console can render the "all clear" baseline (ADR-019) and apply only deltas
/// with `cluster_seq > snapshot.as_of_seq`.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StreamedClusterSnapshot {
    /// Frame discriminator pinned to `"cluster_snapshot"`.
    pub kind: String,
    /// The calm-state baseline.
    pub snapshot: aion_core::ClusterSnapshot,
}

impl StreamedClusterSnapshot {
    /// Frame discriminator value for the priming snapshot.
    pub const KIND: &'static str = "cluster_snapshot";

    /// Wrap a snapshot in its server->client priming frame.
    #[must_use]
    pub fn new(snapshot: aion_core::ClusterSnapshot) -> Self {
        Self {
            kind: Self::KIND.to_owned(),
            snapshot,
        }
    }
}

/// Subscribe to the agent-observability transcript for one
/// `(workflow, run, activity, attempt)` (NOI-5b).
///
/// Namespace-scoped: the transcript belongs to the workflow the activity runs
/// under, so this carries the same `namespace` + `workflow_id` the per-workflow
/// event subscription does, plus the `run_id`/`activity_id`/`attempt` axes that
/// pin the exact `O`-keyspace stream. The optional `after_seq` resume cursor is the
/// highest `store_seq` the client has already applied; the server replays the
/// durable `O` tail with `store_seq > after_seq` then splices onto the live
/// broadcast with no gap and no duplicate (the same splice contract the workflow
/// path's `resume_from_seq` uses, but keyed on the commit-allocated `store_seq`).
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct TranscriptSubscription {
    /// Caller namespace used for adapter-boundary authorisation (the workflow's
    /// namespace, identical to the per-workflow event subscription).
    #[prost(string, tag = "1")]
    pub namespace: String,
    /// Workflow whose activity transcript is requested.
    #[prost(message, optional, tag = "2")]
    pub workflow_id: Option<ProtoWorkflowId>,
    /// Concrete run of that workflow whose activity transcript is requested —
    /// the second stream axis, and REQUIRED.
    ///
    /// `Option` here is a prost message-field encoding artefact, exactly as it
    /// is for `workflow_id`: the server rejects an absent value as
    /// `invalid_input` rather than defaulting it. There is no "latest run"
    /// fallback, because a transcript subscription that silently resolved to a
    /// different generation than the caller meant is the ambiguity this axis
    /// exists to remove.
    #[prost(message, optional, tag = "6")]
    pub run_id: Option<ProtoRunId>,
    /// Activity within the workflow whose transcript is requested.
    #[prost(message, optional, tag = "3")]
    pub activity_id: Option<crate::convert::ProtoActivityId>,
    /// Attempt number — the third stream axis. Two attempts of one activity are
    /// DISTINCT transcript streams.
    #[prost(uint32, tag = "4")]
    pub attempt: u32,
    /// Highest `store_seq` already applied by the client; the server suppresses
    /// durable records and live deltas with `store_seq <= after_seq` so a
    /// reconnect does not re-deliver them. Absent (`None`) = a fresh subscriber
    /// that has applied nothing and must see the full durable transcript
    /// including `store_seq == 0`.
    #[prost(uint64, optional, tag = "5")]
    pub after_seq: Option<u64>,
}

/// Server -> client frame wrapping a single [`aion_core::ActivityEvent`] on the
/// agent-observability transcript channel (NOI-5b).
///
/// Mirrors [`StreamedClusterEvent`] for the transcript path: the inner
/// `aion-core` type is the only thing that crosses the ts-rs boundary; this
/// outer envelope is hand-decoded on the TS side. A persisted event carries its
/// commit-allocated `store_seq`; an ephemeral token delta carries `store_seq:
/// None` and is forwarded live but never replayed.
#[derive(Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct StreamedActivityEvent {
    /// Frame discriminator pinned to `"activity_event"` so the ops console's
    /// hand-written frame parser can branch a transcript event apart from an
    /// `{"error": ...}` terminal frame.
    pub kind: String,
    /// The transcript event.
    pub event: aion_core::ActivityEvent,
}

impl StreamedActivityEvent {
    /// Frame discriminator value for a live transcript event.
    pub const KIND: &'static str = "activity_event";

    /// Wrap an activity event in its server->client frame.
    #[must_use]
    pub fn new(event: aion_core::ActivityEvent) -> Self {
        Self {
            kind: Self::KIND.to_owned(),
            event,
        }
    }
}

/// Subscribe to events for one workflow.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct PerWorkflowSubscription {
    /// Caller namespace used for adapter-boundary authorisation.
    #[prost(string, tag = "1")]
    pub namespace: String,
    /// Workflow whose events are requested.
    #[prost(message, optional, tag = "2")]
    pub workflow_id: Option<ProtoWorkflowId>,
    /// First per-workflow sequence number the caller wants — not the last seq
    /// already seen. When present, the server replays recorded history events
    /// with seq >= `resume_from_seq` in order, then splices into the live
    /// stream with no gaps and no duplicates. Sequence numbers start at 1; 0
    /// is rejected as `invalid_input`. Absent = live tail only (current
    /// behaviour). `resume_from_seq` = 1 replays the full history.
    ///
    /// Only per-workflow subscriptions carry a resume cursor: per-workflow
    /// seq is the only ordering that exists, so [`FilteredSubscription`] and
    /// [`FirehoseSubscription`] are live-only by design.
    ///
    /// RESERVED compaction signal (documentation-only, no code yet): a cursor
    /// older than the earliest retained event yields `not_found` with
    /// `error_type` `"HistoryCompacted"`; callers restart with a fresh
    /// subscription.
    #[prost(uint64, optional, tag = "3")]
    pub resume_from_seq: Option<u64>,
}

/// Subscribe to events selected by optional workflow metadata.
///
/// Filtered streams carry NO resume cursor and are live-only by design:
/// per-workflow seq is the only ordering that exists, so resumption is not
/// representable here. Disconnection after at least one delivered event
/// surfaces Unavailable client-side — never a silent gapped reattach.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct FilteredSubscription {
    /// Caller namespace used for adapter-boundary authorisation.
    #[prost(string, tag = "1")]
    pub namespace: String,
    /// Optional workflow type selector.
    #[prost(string, optional, tag = "2")]
    pub workflow_type: Option<String>,
    /// Optional workflow status selector.
    #[prost(enumeration = "ProtoWorkflowStatus", optional, tag = "3")]
    pub status: Option<i32>,
    /// Optional namespace selector distinct from the caller namespace.
    #[prost(string, optional, tag = "4")]
    pub namespace_selector: Option<String>,
}

/// Subscribe to every event visible in the caller's namespace.
///
/// Firehose streams carry NO resume cursor and are live-only by design:
/// per-workflow seq is the only ordering that exists, so resumption is not
/// representable here. Disconnection after at least one delivered event
/// surfaces Unavailable client-side — never a silent gapped reattach.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct FirehoseSubscription {
    /// Caller namespace used for adapter-boundary authorisation.
    #[prost(string, tag = "1")]
    pub namespace: String,
}

/// Streamed event frame carrying an unmodified aion-core `Event` in a wire envelope.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct StreamedEvent {
    /// Namespace that owns the event.
    #[prost(string, tag = "1")]
    pub namespace: String,
    /// Serde-encoded aion-core `Event` envelope.
    #[prost(message, optional, tag = "2")]
    pub event: Option<WireEnvelope>,
}

impl StreamedEvent {
    /// Serializes an aion-core event into a streamed event frame.
    ///
    /// # Errors
    ///
    /// Returns [`WireError`] with code `backend` if the event cannot be
    /// serialized into the shared core-value envelope.
    pub fn encode(
        namespace: impl Into<String>,
        request_id: Option<String>,
        event: &aion_core::Event,
    ) -> Result<Self, WireError> {
        let namespace = namespace.into();
        let event = encode_core_value(namespace.clone(), request_id, event)?;
        Ok(Self {
            namespace,
            event: Some(event),
        })
    }

    /// Decodes the enclosed aion-core event after checking namespace consistency.
    ///
    /// # Errors
    ///
    /// Returns [`WireError`] with code `backend` if the frame is missing its
    /// event envelope, if the frame namespace differs from the envelope
    /// namespace, or if the core event cannot be decoded.
    pub fn decode_event(&self) -> Result<aion_core::Event, WireError> {
        let event = self
            .event
            .as_ref()
            .ok_or_else(|| WireError::backend("streamed event envelope is missing"))?;
        if event.namespace != self.namespace {
            return Err(WireError::backend("streamed event namespace mismatch"));
        }
        decode_core_value(event)
    }
}

/// Serializes an aion-core event into a streamed event frame.
///
/// # Errors
///
/// Returns [`WireError`] with code `backend` if the event cannot be serialized.
pub fn encode_streamed_event(
    namespace: impl Into<String>,
    request_id: Option<String>,
    event: &aion_core::Event,
) -> Result<StreamedEvent, WireError> {
    StreamedEvent::encode(namespace, request_id, event)
}

#[cfg(test)]
mod tests {
    use chrono::{DateTime, Utc};
    use prost::Message;
    use serde_json::json;

    use super::{
        FilteredSubscription, FirehoseSubscription, PerWorkflowSubscription, StreamedEvent,
        SubscriptionRequest, TranscriptSubscription, encode_streamed_event, subscription_request,
    };
    use crate::convert::{
        ProtoActivityId, ProtoRunId, ProtoWorkflowId, ProtoWorkflowStatus, WireEnvelope,
    };
    use crate::error::WireError;

    fn workflow_id() -> aion_core::WorkflowId {
        aion_core::WorkflowId::new(uuid::Uuid::nil())
    }

    fn recorded_at() -> Result<DateTime<Utc>, chrono::ParseError> {
        Ok(DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?.with_timezone(&Utc))
    }

    fn event_envelope() -> Result<aion_core::EventEnvelope, chrono::ParseError> {
        Ok(aion_core::EventEnvelope {
            seq: 1,
            recorded_at: recorded_at()?,
            workflow_id: workflow_id(),
        })
    }

    #[test]
    fn subscription_request_round_trips_all_variants() -> Result<(), Box<dyn std::error::Error>> {
        let requests = [
            SubscriptionRequest {
                subscription: Some(subscription_request::Subscription::PerWorkflow(
                    PerWorkflowSubscription {
                        namespace: String::from("tenant-a"),
                        workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
                        resume_from_seq: None,
                    },
                )),
            },
            SubscriptionRequest {
                subscription: Some(subscription_request::Subscription::PerWorkflow(
                    PerWorkflowSubscription {
                        namespace: String::from("tenant-a"),
                        workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
                        resume_from_seq: Some(42),
                    },
                )),
            },
            SubscriptionRequest {
                subscription: Some(subscription_request::Subscription::Filtered(
                    FilteredSubscription {
                        namespace: String::from("tenant-a"),
                        workflow_type: Some(String::from("checkout")),
                        status: Some(ProtoWorkflowStatus::Running as i32),
                        namespace_selector: Some(String::from("tenant-a")),
                    },
                )),
            },
            SubscriptionRequest {
                subscription: Some(subscription_request::Subscription::Filtered(
                    FilteredSubscription {
                        namespace: String::from("tenant-a"),
                        workflow_type: None,
                        status: None,
                        namespace_selector: None,
                    },
                )),
            },
            SubscriptionRequest {
                subscription: Some(subscription_request::Subscription::Firehose(
                    FirehoseSubscription {
                        namespace: String::from("tenant-a"),
                    },
                )),
            },
            SubscriptionRequest {
                subscription: Some(subscription_request::Subscription::Transcript(
                    TranscriptSubscription {
                        namespace: String::from("tenant-a"),
                        workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
                        run_id: Some(ProtoRunId::from(aion_core::RunId::new(
                            uuid::Uuid::from_u128(0x11),
                        ))),
                        activity_id: Some(ProtoActivityId {
                            sequence_position: 3,
                        }),
                        attempt: 1,
                        after_seq: Some(9),
                    },
                )),
            },
            SubscriptionRequest {
                subscription: Some(subscription_request::Subscription::Transcript(
                    TranscriptSubscription {
                        namespace: String::from("tenant-a"),
                        workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
                        run_id: Some(ProtoRunId::from(aion_core::RunId::new(
                            uuid::Uuid::from_u128(0x22),
                        ))),
                        activity_id: Some(ProtoActivityId {
                            sequence_position: 3,
                        }),
                        attempt: 0,
                        after_seq: None,
                    },
                )),
            },
        ];

        for request in requests {
            let json = serde_json::to_vec(&request)?;
            let from_json: SubscriptionRequest = serde_json::from_slice(&json)?;
            assert_eq!(from_json, request);

            let bytes = request.encode_to_vec();
            let from_proto = SubscriptionRequest::decode(bytes.as_slice())?;
            assert_eq!(from_proto, request);
        }

        Ok(())
    }

    #[test]
    fn per_workflow_resume_cursor_round_trips_prost() -> Result<(), Box<dyn std::error::Error>> {
        let with_cursor = PerWorkflowSubscription {
            namespace: String::from("tenant-a"),
            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
            resume_from_seq: Some(7),
        };
        let decoded = PerWorkflowSubscription::decode(with_cursor.encode_to_vec().as_slice())?;
        assert_eq!(decoded, with_cursor);
        assert_eq!(decoded.resume_from_seq, Some(7));

        let without_cursor = PerWorkflowSubscription {
            namespace: String::from("tenant-a"),
            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
            resume_from_seq: None,
        };
        let decoded = PerWorkflowSubscription::decode(without_cursor.encode_to_vec().as_slice())?;
        assert_eq!(decoded, without_cursor);
        assert_eq!(decoded.resume_from_seq, None);

        Ok(())
    }

    #[test]
    fn per_workflow_resume_cursor_json_shape_is_pinned() -> Result<(), Box<dyn std::error::Error>> {
        let with_cursor = PerWorkflowSubscription {
            namespace: String::from("tenant-a"),
            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
            resume_from_seq: Some(7),
        };
        let value = serde_json::to_value(&with_cursor)?;
        assert_eq!(
            value,
            json!({
                "namespace": "tenant-a",
                "workflow_id": { "uuid": "00000000-0000-0000-0000-000000000000" },
                "resume_from_seq": 7,
            })
        );
        let from_json: PerWorkflowSubscription = serde_json::from_value(value)?;
        assert_eq!(from_json, with_cursor);

        let without_cursor = PerWorkflowSubscription {
            namespace: String::from("tenant-a"),
            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
            resume_from_seq: None,
        };
        let value = serde_json::to_value(&without_cursor)?;
        assert_eq!(
            value,
            json!({
                "namespace": "tenant-a",
                "workflow_id": { "uuid": "00000000-0000-0000-0000-000000000000" },
                "resume_from_seq": null,
            })
        );
        let from_json: PerWorkflowSubscription = serde_json::from_value(value)?;
        assert_eq!(from_json, without_cursor);

        Ok(())
    }

    #[test]
    fn subscription_request_without_resume_field_decodes_to_none()
    -> Result<(), Box<dyn std::error::Error>> {
        let request: SubscriptionRequest = serde_json::from_value(json!({
            "subscription": {
                "PerWorkflow": {
                    "namespace": "tenant-a",
                    "workflow_id": { "uuid": "00000000-0000-0000-0000-000000000000" },
                }
            }
        }))?;

        let Some(subscription_request::Subscription::PerWorkflow(per_workflow)) =
            request.subscription
        else {
            return Err(Box::from("expected a per-workflow subscription"));
        };
        assert_eq!(per_workflow.namespace, "tenant-a");
        assert_eq!(
            per_workflow.workflow_id,
            Some(ProtoWorkflowId::from(workflow_id()))
        );
        assert_eq!(per_workflow.resume_from_seq, None);

        Ok(())
    }

    #[test]
    fn streamed_event_round_trips_core_event() -> Result<(), Box<dyn std::error::Error>> {
        let event = aion_core::Event::WorkflowStarted {
            envelope: event_envelope()?,
            workflow_type: String::from("checkout"),
            input: aion_core::Payload::from_json(&json!({ "cart": ["sku-1"] }))?,
            run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
            parent_run_id: None,
            parent_workflow_id: None,
            package_version: aion_core::PackageVersion::new("a".repeat(64)),
        };

        let frame = encode_streamed_event("tenant-a", Some(String::from("request-1")), &event)?;
        assert_eq!(frame.namespace, "tenant-a");
        let envelope = frame
            .event
            .as_ref()
            .ok_or_else(|| WireError::backend("test streamed event envelope is missing"))?;
        assert_eq!(envelope.namespace, "tenant-a");
        assert_eq!(envelope.request_id.as_deref(), Some("request-1"));

        let decoded = frame.decode_event()?;
        assert_eq!(decoded, event);
        Ok(())
    }

    #[test]
    fn streamed_event_rejects_namespace_mismatch() {
        let frame = StreamedEvent {
            namespace: String::from("tenant-a"),
            event: Some(WireEnvelope {
                namespace: String::from("tenant-b"),
                request_id: None,
                payload: None,
            }),
        };

        assert_eq!(
            frame.decode_event(),
            Err(WireError::backend("streamed event namespace mismatch"))
        );
    }

    #[test]
    fn streamed_event_rejects_missing_envelope() {
        let frame = StreamedEvent {
            namespace: String::from("tenant-a"),
            event: None,
        };

        assert_eq!(
            frame.decode_event(),
            Err(WireError::backend("streamed event envelope is missing"))
        );
    }
}