lsp_max_protocol/
stream.rs

1use serde::{Deserialize, Serialize};
2
3// max/stream.subscribe
4pub const STREAM_SUBSCRIBE: &str = "max/stream.subscribe";
5// max/stream.unsubscribe
6pub const STREAM_UNSUBSCRIBE: &str = "max/stream.unsubscribe";
7// max/stream.event — pushed from server to client
8pub const STREAM_EVENT: &str = "max/stream.event";
9
10#[derive(Debug, Clone, Serialize, Deserialize)]
11pub struct StreamSubscribeParams {
12    /// Subscription ID chosen by the client
13    pub subscription_id: String,
14    /// Which event kinds to subscribe to
15    pub event_kinds: Vec<StreamEventKind>,
16    /// Optional: only events for this document URI (serialised URI string)
17    pub uri_filter: Option<String>,
18}
19
20#[derive(Debug, Clone, Serialize, Deserialize)]
21pub struct StreamSubscribeResult {
22    pub subscription_id: String,
23    /// "CANDIDATE" — streaming is active but not ADMITTED
24    pub status: String,
25}
26
27#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct StreamUnsubscribeParams {
29    pub subscription_id: String,
30}
31
32#[derive(Debug, Clone, Serialize, Deserialize)]
33pub struct StreamUnsubscribeResult {
34    pub subscription_id: String,
35    pub events_received: u64,
36    pub status: String,
37}
38
39#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
40pub enum StreamEventKind {
41    Diagnostic,
42    ConformanceChange,
43    GateChange,
44    ReceiptAdmission,
45    LawViolation,
46    /// Admission gate outcome for an agent operation request.
47    /// Payload: BrokerDecisionPayload
48    BrokerDecision,
49    /// Agent's position in the admission queue changed.
50    /// Payload: QueuePositionPayload
51    QueuePositionChange,
52}
53
54/// Admission decision variants pushed via BrokerDecision events.
55#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
56#[serde(tag = "kind", rename_all = "SCREAMING_SNAKE_CASE")]
57pub enum AdmissionDecision {
58    Admitted,
59    Refused,
60    Queued { position: u32 },
61}
62
63/// Payload for StreamEventKind::BrokerDecision.
64#[derive(Debug, Clone, Serialize, Deserialize)]
65pub struct BrokerDecisionPayload {
66    /// Operation ID, e.g. "ggen.sync", "ggen.build.workspace"
67    pub op: String,
68    pub decision: AdmissionDecision,
69    /// BLAKE3 receipt hash if admitted
70    pub receipt_hash: Option<String>,
71    /// DefectClass name if refused
72    pub refusal_reason: Option<String>,
73}
74
75impl BrokerDecisionPayload {
76    pub fn admitted(op: impl Into<String>, receipt_hash: impl Into<String>) -> Self {
77        Self {
78            op: op.into(),
79            decision: AdmissionDecision::Admitted,
80            receipt_hash: Some(receipt_hash.into()),
81            refusal_reason: None,
82        }
83    }
84
85    pub fn refused(op: impl Into<String>, reason: impl Into<String>) -> Self {
86        Self {
87            op: op.into(),
88            decision: AdmissionDecision::Refused,
89            receipt_hash: None,
90            refusal_reason: Some(reason.into()),
91        }
92    }
93
94    pub fn queued(op: impl Into<String>, position: u32) -> Self {
95        Self {
96            op: op.into(),
97            decision: AdmissionDecision::Queued { position },
98            receipt_hash: None,
99            refusal_reason: None,
100        }
101    }
102}
103
104/// Payload for StreamEventKind::QueuePositionChange.
105#[derive(Debug, Clone, Serialize, Deserialize)]
106pub struct QueuePositionPayload {
107    pub op: String,
108    pub position: u32,
109    pub queue_depth: u32,
110}
111
112#[derive(Debug, Clone, Serialize, Deserialize)]
113pub struct StreamEvent {
114    pub subscription_id: String,
115    pub sequence: u64,
116    pub kind: StreamEventKind,
117    pub payload: serde_json::Value,
118    /// Law-axis status of this event's source.
119    pub status: String,
120    pub timestamp_secs: u64,
121}
122
123/// In-process event bus for max/stream subscriptions.
124/// Backed by tokio::sync::broadcast.
125pub struct StreamBus {
126    sender: tokio::sync::broadcast::Sender<StreamEvent>,
127}
128
129impl StreamBus {
130    pub fn new(capacity: usize) -> Self {
131        let (sender, _) = tokio::sync::broadcast::channel(capacity);
132        Self { sender }
133    }
134
135    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<StreamEvent> {
136        self.sender.subscribe()
137    }
138
139    /// Publish an event to all current subscribers. Returns the receiver count.
140    pub fn publish(&self, event: StreamEvent) -> usize {
141        self.sender.send(event).unwrap_or(0)
142    }
143
144    pub fn subscriber_count(&self) -> usize {
145        self.sender.receiver_count()
146    }
147}
lsp_max_protocol/stream.rs

lsp_max_protocol/
stream.rs
