diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 12ba8d8a..fe33ca35 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -1,5 +1,6 @@ #[cfg(feature = "stream")] pub mod stream; +pub mod subscription; #[cfg(feature = "typescript")] pub mod typescript; diff --git a/crates/protocol/src/subscription.rs b/crates/protocol/src/subscription.rs new file mode 100644 index 00000000..b2d9352e --- /dev/null +++ b/crates/protocol/src/subscription.rs @@ -0,0 +1,908 @@ +//! Shared subscription-multiplexer wire contract. +//! +//! This protocol is distinct from the single-Worker [`crate::Method`] / [`crate::Event`] +//! protocol. A connection carries many independently managed subscriptions. The +//! connection endpoint owns authorization and maps the typed selectors below to +//! visible resources; clients cannot provide Workspace scope as selector input. + +use std::collections::HashSet; +use std::fmt; + +use serde::{Deserialize, Deserializer, Serialize}; + +use crate::Event as WorkerProtocolEvent; + +pub const SUBSCRIPTION_PROTOCOL_VERSION: u16 = 1; +pub const MAX_CORRELATION_ID_BYTES: usize = 128; +pub const MAX_RESOURCE_ID_BYTES: usize = 256; +pub const MAX_WORKER_IDS_PER_SELECTOR: usize = 256; +pub const MAX_REJECTION_MESSAGE_BYTES: usize = 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SubscriptionValidationError { + UnsupportedProtocolVersion { + actual: u16, + }, + EmptyIdentifier { + field: &'static str, + }, + IdentifierTooLong { + field: &'static str, + max_bytes: usize, + }, + InvalidIdentifier { + field: &'static str, + }, + EmptyWorkerSelection, + TooManyWorkers { + max: usize, + }, + DuplicateWorkerId { + worker_id: String, + }, + EmptyRejectionMessage, + RejectionMessageTooLong { + max_bytes: usize, + }, + SelectorSnapshotMismatch, + SelectorEventMismatch, + UnselectedWorker { + worker_id: String, + }, +} + +impl fmt::Display for SubscriptionValidationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::UnsupportedProtocolVersion { actual } => write!( + formatter, + "unsupported subscription protocol version {actual}; expected {SUBSCRIPTION_PROTOCOL_VERSION}" + ), + Self::EmptyIdentifier { field } => write!(formatter, "{field} must not be empty"), + Self::IdentifierTooLong { field, max_bytes } => { + write!(formatter, "{field} exceeds {max_bytes} bytes") + } + Self::InvalidIdentifier { field } => write!( + formatter, + "{field} must be trimmed and must not contain control characters" + ), + Self::EmptyWorkerSelection => { + formatter.write_str("worker_lifecycle requires at least one worker id") + } + Self::TooManyWorkers { max } => { + write!(formatter, "worker selector exceeds {max} worker ids") + } + Self::DuplicateWorkerId { worker_id } => { + write!( + formatter, + "worker selector contains duplicate id {worker_id:?}" + ) + } + Self::EmptyRejectionMessage => { + formatter.write_str("subscription rejection message must not be empty") + } + Self::RejectionMessageTooLong { max_bytes } => write!( + formatter, + "subscription rejection message exceeds {max_bytes} bytes" + ), + Self::SelectorSnapshotMismatch => { + formatter.write_str("subscription snapshot does not match its selector") + } + Self::SelectorEventMismatch => { + formatter.write_str("subscription event does not match its selector") + } + Self::UnselectedWorker { worker_id } => { + write!(formatter, "worker {worker_id:?} is outside the selector") + } + } + } +} + +impl std::error::Error for SubscriptionValidationError {} + +macro_rules! bounded_identifier { + ($name:ident, $field:literal, $max:expr) => { + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] + #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] + pub struct $name(String); + + impl $name { + pub fn new(value: impl Into) -> Result { + let value = value.into(); + validate_identifier($field, &value, $max)?; + Ok(Self(value)) + } + + pub fn as_str(&self) -> &str { + &self.0 + } + + pub fn into_inner(self) -> String { + self.0 + } + + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + validate_identifier($field, &self.0, $max) + } + } + + impl AsRef for $name { + fn as_ref(&self) -> &str { + self.as_str() + } + } + + impl fmt::Display for $name { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.as_str()) + } + } + }; +} + +bounded_identifier!( + SubscriptionRequestId, + "request_id", + MAX_CORRELATION_ID_BYTES +); +bounded_identifier!(SubscriptionId, "subscription_id", MAX_CORRELATION_ID_BYTES); +bounded_identifier!(SubscriptionWorkerId, "worker_id", MAX_RESOURCE_ID_BYTES); +bounded_identifier!( + SubscriptionWorkdirId, + "working_directory_id", + MAX_RESOURCE_ID_BYTES +); + +fn validate_identifier( + field: &'static str, + value: &str, + max_bytes: usize, +) -> Result<(), SubscriptionValidationError> { + if value.is_empty() { + return Err(SubscriptionValidationError::EmptyIdentifier { field }); + } + if value.len() > max_bytes { + return Err(SubscriptionValidationError::IdentifierTooLong { field, max_bytes }); + } + if value.trim() != value || value.chars().any(char::is_control) { + return Err(SubscriptionValidationError::InvalidIdentifier { field }); + } + Ok(()) +} + +fn validate_rejection_message(message: &str) -> Result<(), SubscriptionValidationError> { + if message.is_empty() { + return Err(SubscriptionValidationError::EmptyRejectionMessage); + } + if message.len() > MAX_REJECTION_MESSAGE_BYTES { + return Err(SubscriptionValidationError::RejectionMessageTooLong { + max_bytes: MAX_REJECTION_MESSAGE_BYTES, + }); + } + Ok(()) +} + +/// A bounded, canonical worker-id set. Ordering on the wire and in equality/hash +/// semantics is lexical, so equivalent selectors aggregate to one upstream key. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct SubscriptionWorkerIds(Vec); + +impl SubscriptionWorkerIds { + pub fn new( + worker_ids: impl IntoIterator, + ) -> Result { + let mut worker_ids = worker_ids.into_iter().collect::>(); + if worker_ids.is_empty() { + return Err(SubscriptionValidationError::EmptyWorkerSelection); + } + if worker_ids.len() > MAX_WORKER_IDS_PER_SELECTOR { + return Err(SubscriptionValidationError::TooManyWorkers { + max: MAX_WORKER_IDS_PER_SELECTOR, + }); + } + for worker_id in &worker_ids { + worker_id.validate()?; + } + worker_ids.sort_unstable(); + for pair in worker_ids.windows(2) { + if pair[0] == pair[1] { + return Err(SubscriptionValidationError::DuplicateWorkerId { + worker_id: pair[0].to_string(), + }); + } + } + Ok(Self(worker_ids)) + } + + pub fn as_slice(&self) -> &[SubscriptionWorkerId] { + &self.0 + } + + pub fn contains(&self, worker_id: &SubscriptionWorkerId) -> bool { + self.0.binary_search(worker_id).is_ok() + } + + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + Self::new(self.0.clone()).map(|_| ()) + } +} + +impl<'de> Deserialize<'de> for SubscriptionWorkerIds { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let worker_ids = Vec::::deserialize(deserializer)?; + Self::new(worker_ids).map_err(serde::de::Error::custom) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "topic", rename_all = "snake_case")] +pub enum EventSubscriptionSelector { + RuntimeWorkers, + WorkerLifecycle { + worker_ids: SubscriptionWorkerIds, + }, + WorkerProtocol { + worker_id: SubscriptionWorkerId, + }, + /// Server-derived Workspace projection. Workspace identity comes from the + /// authenticated connection and is deliberately absent from this selector. + WorkspaceWorkers, + /// Typed extension point for the later Workdir subscription slice. + WorkspaceWorkdirs, +} + +impl EventSubscriptionSelector { + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + match self { + Self::WorkerLifecycle { worker_ids } => worker_ids.validate(), + Self::WorkerProtocol { worker_id } => worker_id.validate(), + Self::RuntimeWorkers | Self::WorkspaceWorkers | Self::WorkspaceWorkdirs => Ok(()), + } + } + + pub fn selects_worker(&self, worker_id: &SubscriptionWorkerId) -> bool { + match self { + Self::RuntimeWorkers | Self::WorkspaceWorkers => true, + Self::WorkerLifecycle { worker_ids } => worker_ids.contains(worker_id), + Self::WorkerProtocol { + worker_id: selected, + } => selected == worker_id, + Self::WorkspaceWorkdirs => false, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct SubscriptionFrame { + pub protocol_version: u16, + #[serde(flatten)] + pub payload: SubscriptionFramePayload, +} + +impl SubscriptionFrame { + pub fn new(payload: SubscriptionFramePayload) -> Self { + Self { + protocol_version: SUBSCRIPTION_PROTOCOL_VERSION, + payload, + } + } + + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + if self.protocol_version != SUBSCRIPTION_PROTOCOL_VERSION { + return Err(SubscriptionValidationError::UnsupportedProtocolVersion { + actual: self.protocol_version, + }); + } + self.payload.validate() + } +} + +/// Direction is explicit at the outer envelope, while concrete message kinds +/// remain typed within each lane. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "frame", content = "message", rename_all = "snake_case")] +pub enum SubscriptionFramePayload { + Request(SubscriptionRequest), + Response(SubscriptionResponse), + Event(SubscriptionEvent), +} + +impl SubscriptionFramePayload { + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + match self { + Self::Request(request) => request.validate(), + Self::Response(response) => response.validate(), + Self::Event(event) => event.validate(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "method", content = "params", rename_all = "snake_case")] +pub enum SubscriptionRequest { + SubscribeEvents { + request_id: SubscriptionRequestId, + selector: EventSubscriptionSelector, + }, + UnsubscribeEvents { + request_id: SubscriptionRequestId, + subscription_id: SubscriptionId, + }, +} + +impl SubscriptionRequest { + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + match self { + Self::SubscribeEvents { + request_id, + selector, + } => { + request_id.validate()?; + selector.validate() + } + Self::UnsubscribeEvents { + request_id, + subscription_id, + } => { + request_id.validate()?; + subscription_id.validate() + } + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "result", content = "payload", rename_all = "snake_case")] +pub enum SubscriptionResponse { + Subscribed { + request_id: SubscriptionRequestId, + subscription_id: SubscriptionId, + selector: EventSubscriptionSelector, + snapshot_revision: u64, + snapshot: SubscriptionSnapshot, + }, + Unsubscribed { + request_id: SubscriptionRequestId, + subscription_id: SubscriptionId, + }, + SubscriptionRejected { + request_id: SubscriptionRequestId, + #[serde(default, skip_serializing_if = "Option::is_none")] + subscription_id: Option, + code: SubscriptionRejectionCode, + message: String, + }, +} + +impl SubscriptionResponse { + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + match self { + Self::Subscribed { + request_id, + subscription_id, + selector, + snapshot, + .. + } => { + request_id.validate()?; + subscription_id.validate()?; + selector.validate()?; + snapshot.validate_for_selector(selector) + } + Self::Unsubscribed { + request_id, + subscription_id, + } => { + request_id.validate()?; + subscription_id.validate() + } + Self::SubscriptionRejected { + request_id, + subscription_id, + message, + .. + } => { + request_id.validate()?; + if let Some(subscription_id) = subscription_id { + subscription_id.validate()?; + } + validate_rejection_message(message) + } + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum SubscriptionRejectionCode { + InvalidRequest, + UnsupportedProtocolVersion, + UnsupportedSelector, + Unauthorized, + ResourceNotFound, + CapacityExceeded, + Internal, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum SubscriptionTerminationCode { + Lagged, + ResourceGone, + Unauthorized, + ServerShutdown, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "event", content = "data", rename_all = "snake_case")] +pub enum SubscriptionEvent { + Event { + subscription_id: SubscriptionId, + subject_revision: u64, + payload: SubscriptionEventPayload, + }, + SubscriptionClosed { + subscription_id: SubscriptionId, + code: SubscriptionTerminationCode, + message: String, + }, +} + +impl SubscriptionEvent { + pub fn subscription_id(&self) -> &SubscriptionId { + match self { + Self::Event { + subscription_id, .. + } + | Self::SubscriptionClosed { + subscription_id, .. + } => subscription_id, + } + } + + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + match self { + Self::Event { + subscription_id, + payload, + .. + } => { + subscription_id.validate()?; + payload.validate() + } + Self::SubscriptionClosed { + subscription_id, + message, + .. + } => { + subscription_id.validate()?; + validate_rejection_message(message) + } + } + } + + pub fn validate_for_selector( + &self, + selector: &EventSubscriptionSelector, + ) -> Result<(), SubscriptionValidationError> { + self.validate()?; + match self { + Self::Event { payload, .. } => payload.validate_for_selector(selector), + Self::SubscriptionClosed { .. } => Ok(()), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum SubscriptionWorkerState { + Idle, + Running, + Paused, + Stopped, + Cancelled, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct SubscriptionWorker { + pub worker_id: SubscriptionWorkerId, + /// Producer-owned monotonic revision for this Worker subject. + pub subject_revision: u64, + pub state: SubscriptionWorkerState, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub profile: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory_id: Option, +} + +impl SubscriptionWorker { + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + self.worker_id.validate()?; + if let Some(working_directory_id) = &self.working_directory_id { + working_directory_id.validate()?; + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct SubscriptionWorkdir { + pub working_directory_id: SubscriptionWorkdirId, + pub repository_id: String, + pub state: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub primary_worker_id: Option, +} + +impl SubscriptionWorkdir { + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + self.working_directory_id.validate()?; + validate_identifier("repository_id", &self.repository_id, MAX_RESOURCE_ID_BYTES)?; + validate_identifier("workdir_state", &self.state, MAX_RESOURCE_ID_BYTES)?; + if let Some(worker_id) = &self.primary_worker_id { + worker_id.validate()?; + } + Ok(()) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "topic", content = "data", rename_all = "snake_case")] +pub enum SubscriptionSnapshot { + Workers { + workers: Vec, + }, + WorkerProtocol { + worker_id: SubscriptionWorkerId, + events: Vec, + }, + WorkspaceWorkdirs { + workdirs: Vec, + }, +} + +impl SubscriptionSnapshot { + pub fn validate_for_selector( + &self, + selector: &EventSubscriptionSelector, + ) -> Result<(), SubscriptionValidationError> { + selector.validate()?; + match (selector, self) { + ( + EventSubscriptionSelector::RuntimeWorkers + | EventSubscriptionSelector::WorkspaceWorkers, + Self::Workers { workers }, + ) => validate_workers(workers), + ( + EventSubscriptionSelector::WorkerLifecycle { worker_ids }, + Self::Workers { workers }, + ) => { + validate_workers(workers)?; + for worker in workers { + if !worker_ids.contains(&worker.worker_id) { + return Err(SubscriptionValidationError::UnselectedWorker { + worker_id: worker.worker_id.to_string(), + }); + } + } + Ok(()) + } + ( + EventSubscriptionSelector::WorkerProtocol { + worker_id: selected, + }, + Self::WorkerProtocol { worker_id, .. }, + ) if selected == worker_id => worker_id.validate(), + ( + EventSubscriptionSelector::WorkspaceWorkdirs, + Self::WorkspaceWorkdirs { workdirs }, + ) => { + for workdir in workdirs { + workdir.validate()?; + } + Ok(()) + } + _ => Err(SubscriptionValidationError::SelectorSnapshotMismatch), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "event", content = "data", rename_all = "snake_case")] +pub enum SubscriptionEventPayload { + WorkerUpserted { + worker: SubscriptionWorker, + }, + WorkerRemoved { + worker_id: SubscriptionWorkerId, + }, + WorkerProtocol { + worker_id: SubscriptionWorkerId, + event: WorkerProtocolEvent, + }, + WorkdirUpserted { + workdir: SubscriptionWorkdir, + }, + WorkdirRemoved { + working_directory_id: SubscriptionWorkdirId, + }, +} + +impl SubscriptionEventPayload { + pub fn validate(&self) -> Result<(), SubscriptionValidationError> { + match self { + Self::WorkerUpserted { worker } => worker.validate(), + Self::WorkerRemoved { worker_id } | Self::WorkerProtocol { worker_id, .. } => { + worker_id.validate() + } + Self::WorkdirUpserted { workdir } => workdir.validate(), + Self::WorkdirRemoved { + working_directory_id, + } => working_directory_id.validate(), + } + } + + pub fn validate_for_selector( + &self, + selector: &EventSubscriptionSelector, + ) -> Result<(), SubscriptionValidationError> { + selector.validate()?; + self.validate()?; + match (selector, self) { + ( + EventSubscriptionSelector::RuntimeWorkers + | EventSubscriptionSelector::WorkspaceWorkers, + Self::WorkerUpserted { .. } | Self::WorkerRemoved { .. }, + ) => Ok(()), + ( + EventSubscriptionSelector::WorkerLifecycle { worker_ids }, + Self::WorkerUpserted { worker }, + ) if worker_ids.contains(&worker.worker_id) => Ok(()), + ( + EventSubscriptionSelector::WorkerLifecycle { worker_ids }, + Self::WorkerRemoved { worker_id }, + ) if worker_ids.contains(worker_id) => Ok(()), + ( + EventSubscriptionSelector::WorkerProtocol { + worker_id: selected, + }, + Self::WorkerProtocol { worker_id, .. }, + ) if selected == worker_id => Ok(()), + ( + EventSubscriptionSelector::WorkspaceWorkdirs, + Self::WorkdirUpserted { .. } | Self::WorkdirRemoved { .. }, + ) => Ok(()), + ( + EventSubscriptionSelector::WorkerLifecycle { .. }, + Self::WorkerUpserted { worker }, + ) => Err(SubscriptionValidationError::UnselectedWorker { + worker_id: worker.worker_id.to_string(), + }), + ( + EventSubscriptionSelector::WorkerLifecycle { .. }, + Self::WorkerRemoved { worker_id }, + ) => Err(SubscriptionValidationError::UnselectedWorker { + worker_id: worker_id.to_string(), + }), + _ => Err(SubscriptionValidationError::SelectorEventMismatch), + } + } +} + +fn validate_workers(workers: &[SubscriptionWorker]) -> Result<(), SubscriptionValidationError> { + let mut seen = HashSet::with_capacity(workers.len()); + for worker in workers { + worker.validate()?; + if !seen.insert(&worker.worker_id) { + return Err(SubscriptionValidationError::DuplicateWorkerId { + worker_id: worker.worker_id.to_string(), + }); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn worker_id(value: &str) -> SubscriptionWorkerId { + SubscriptionWorkerId::new(value).unwrap() + } + + fn request_id() -> SubscriptionRequestId { + SubscriptionRequestId::new("request-1").unwrap() + } + + fn subscription_id() -> SubscriptionId { + SubscriptionId::new("subscription-1").unwrap() + } + + fn worker(value: &str) -> SubscriptionWorker { + SubscriptionWorker { + worker_id: worker_id(value), + subject_revision: 0, + state: SubscriptionWorkerState::Idle, + workspace_id: Some("workspace-1".to_string()), + display_name: Some(format!("Worker {value}")), + profile: Some("builtin:coder".to_string()), + working_directory_id: None, + } + } + + #[test] + fn subscribe_frame_has_stable_versioned_json_shape() { + let frame = SubscriptionFrame::new(SubscriptionFramePayload::Request( + SubscriptionRequest::SubscribeEvents { + request_id: request_id(), + selector: EventSubscriptionSelector::WorkerLifecycle { + worker_ids: SubscriptionWorkerIds::new([ + worker_id("worker-2"), + worker_id("worker-1"), + ]) + .unwrap(), + }, + }, + )); + + let json = serde_json::to_value(&frame).unwrap(); + assert_eq!( + json, + serde_json::json!({ + "protocol_version": 1, + "frame": "request", + "message": { + "method": "subscribe_events", + "params": { + "request_id": "request-1", + "selector": { + "topic": "worker_lifecycle", + "worker_ids": ["worker-1", "worker-2"] + } + } + } + }) + ); + frame.validate().unwrap(); + let decoded: SubscriptionFrame = serde_json::from_value(json).unwrap(); + decoded.validate().unwrap(); + } + + #[test] + fn subscribed_snapshot_round_trips_and_matches_selector() { + let frame = SubscriptionFrame::new(SubscriptionFramePayload::Response( + SubscriptionResponse::Subscribed { + request_id: request_id(), + subscription_id: subscription_id(), + selector: EventSubscriptionSelector::RuntimeWorkers, + snapshot_revision: 7, + snapshot: SubscriptionSnapshot::Workers { + workers: vec![worker("worker-1")], + }, + }, + )); + + frame.validate().unwrap(); + let json = serde_json::to_string(&frame).unwrap(); + let decoded: SubscriptionFrame = serde_json::from_str(&json).unwrap(); + decoded.validate().unwrap(); + } + + #[test] + fn worker_selector_is_canonical_and_rejects_duplicates() { + let first = SubscriptionWorkerIds::new([worker_id("b"), worker_id("a")]).unwrap(); + let second = SubscriptionWorkerIds::new([worker_id("a"), worker_id("b")]).unwrap(); + assert_eq!(first, second); + + let error = SubscriptionWorkerIds::new([worker_id("a"), worker_id("a")]).unwrap_err(); + assert!(matches!( + error, + SubscriptionValidationError::DuplicateWorkerId { .. } + )); + assert!(serde_json::from_str::("[]").is_err()); + } + + #[test] + fn validation_rejects_invalid_identifiers_and_versions() { + assert!(matches!( + SubscriptionRequestId::new(" "), + Err(SubscriptionValidationError::InvalidIdentifier { .. }) + )); + let mut frame = SubscriptionFrame::new(SubscriptionFramePayload::Request( + SubscriptionRequest::UnsubscribeEvents { + request_id: request_id(), + subscription_id: subscription_id(), + }, + )); + frame.protocol_version = 2; + assert!(matches!( + frame.validate(), + Err(SubscriptionValidationError::UnsupportedProtocolVersion { actual: 2 }) + )); + } + + #[test] + fn snapshot_and_events_must_match_selector() { + let selected = EventSubscriptionSelector::WorkerLifecycle { + worker_ids: SubscriptionWorkerIds::new([worker_id("worker-1")]).unwrap(), + }; + let snapshot = SubscriptionSnapshot::Workers { + workers: vec![worker("worker-2")], + }; + assert!(matches!( + snapshot.validate_for_selector(&selected), + Err(SubscriptionValidationError::UnselectedWorker { .. }) + )); + + let event = SubscriptionEvent::Event { + subscription_id: subscription_id(), + subject_revision: 8, + payload: SubscriptionEventPayload::WorkerRemoved { + worker_id: worker_id("worker-2"), + }, + }; + assert!(matches!( + event.validate_for_selector(&selected), + Err(SubscriptionValidationError::UnselectedWorker { .. }) + )); + assert!(matches!( + event.validate_for_selector(&EventSubscriptionSelector::WorkspaceWorkdirs), + Err(SubscriptionValidationError::SelectorEventMismatch) + )); + } + + #[test] + fn subscription_closed_is_a_typed_server_event() { + let frame = SubscriptionFrame::new(SubscriptionFramePayload::Event( + SubscriptionEvent::SubscriptionClosed { + subscription_id: subscription_id(), + code: SubscriptionTerminationCode::Lagged, + message: "resubscribe".to_string(), + }, + )); + frame.validate().unwrap(); + assert_eq!( + serde_json::to_value(frame).unwrap(), + serde_json::json!({ + "protocol_version": 1, + "frame": "event", + "message": { + "event": "subscription_closed", + "data": { + "subscription_id": "subscription-1", + "code": "lagged", + "message": "resubscribe" + } + } + }) + ); + } + + #[test] + fn client_selector_has_no_workspace_scope_field() { + let json = serde_json::to_value(EventSubscriptionSelector::WorkspaceWorkers).unwrap(); + assert_eq!(json, serde_json::json!({ "topic": "workspace_workers" })); + assert!(json.get("workspace_id").is_none()); + } +} diff --git a/crates/protocol/src/typescript.rs b/crates/protocol/src/typescript.rs index 67afd6a3..607fd864 100644 --- a/crates/protocol/src/typescript.rs +++ b/crates/protocol/src/typescript.rs @@ -7,6 +7,13 @@ use crate::{ InFlightBlock, InFlightSnapshot, InFlightToolCallState, InvokeKind, MemoryWorkerEvent, Method, Permission, RewindSummary, RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment, TurnResult, WorkerEvent, WorkerStatus, + subscription::{ + EventSubscriptionSelector, SubscriptionEvent, SubscriptionEventPayload, SubscriptionFrame, + SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode, SubscriptionRequest, + SubscriptionRequestId, SubscriptionResponse, SubscriptionSnapshot, + SubscriptionTerminationCode, SubscriptionWorkdir, SubscriptionWorkdirId, + SubscriptionWorker, SubscriptionWorkerId, SubscriptionWorkerIds, SubscriptionWorkerState, + }, }; const GENERATED_RELATIVE_PATH: &str = "../../web/workspace/src/lib/generated/protocol.ts"; @@ -50,6 +57,24 @@ pub fn generated_protocol_types() -> String { push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index b639802a..c34e3cb1 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -12,294 +12,152 @@ export type WorkerStatus = "idle" | "running" | "paused"; export type TurnResult = "finished" | "paused"; -export type InvokeKind = - | "user_send" - | "notify" - | "worker_event" - | "system_reminder" - | "wakeup"; +export type InvokeKind = "user_send" | "notify" | "worker_event" | "system_reminder" | "wakeup"; export type RunResult = "finished" | "paused" | "limit_reached" | "rolled_back"; -export type ErrorCode = - | "already_running" - | "not_running" - | "not_paused" - | "provider_error" - | "tool_error" - | "invalid_request" - | "internal"; +export type ErrorCode = "already_running" | "not_running" | "not_paused" | "provider_error" | "tool_error" | "invalid_request" | "internal"; export type Permission = "read" | "write"; export type InFlightToolCallState = "pending" | "streaming_args" | "done"; export type ScopeRule = { - /** - * Target path. Must be absolute by the time a `Scope` is built from - * this rule — relative paths are resolved per-layer against the - * manifest file's directory (cwd for overlay layers) before cascade - * merge. - */ - target: string; - /** - * Permission level this rule grants (allow) or caps strictly below - * (deny). - */ - permission: Permission; - /** - * When `false`, the rule only matches the target itself and its - * direct children. Defaults to `true`. - */ - recursive: boolean; -}; +/** + * Target path. Must be absolute by the time a `Scope` is built from + * this rule — relative paths are resolved per-layer against the + * manifest file's directory (cwd for overlay layers) before cascade + * merge. + */ +target: string, +/** + * Permission level this rule grants (allow) or caps strictly below + * (deny). + */ +permission: Permission, +/** + * When `false`, the rule only matches the target itself and its + * direct children. Defaults to `true`. + */ +recursive: boolean, }; -export type CompletionEntry = { value: string; is_dir: boolean }; +export type CompletionEntry = { value: string, is_dir: boolean, }; -export type RewindTargetId = { - segment_id: string; - user_input_entry_index: number; -}; +export type RewindTargetId = { segment_id: string, user_input_entry_index: number, }; -export type RewindTarget = { - id: RewindTargetId; - expected_head_entries: number; - truncate_entries: number; - turn_index: number; - timestamp_ms: number | null; - preview: string; - eligible: boolean; - disabled_reason: string | null; - warning: string | null; -}; +export type RewindTarget = { id: RewindTargetId, expected_head_entries: number, truncate_entries: number, turn_index: number, timestamp_ms: number | null, preview: string, eligible: boolean, disabled_reason: string | null, warning: string | null, }; -export type RewindSummary = { - truncated_to_entries: number; - discarded_entries: number; - tool_side_effect_warning: boolean; -}; +export type RewindSummary = { truncated_to_entries: number, discarded_entries: number, tool_side_effect_warning: boolean, }; -export type InFlightBlock = - | { "kind": "text"; text: string; finished?: boolean } - | { "kind": "thinking"; text: string; finished?: boolean } - | { - "kind": "tool_call"; - id: string; - name: string; - args: string; - state?: InFlightToolCallState; - }; +export type InFlightBlock = { "kind": "text", text: string, finished?: boolean, } | { "kind": "thinking", text: string, finished?: boolean, } | { "kind": "tool_call", id: string, name: string, args: string, state?: InFlightToolCallState, }; -export type InFlightSnapshot = { blocks?: Array }; +export type InFlightSnapshot = { blocks?: Array, }; -export type Greeting = { - worker_name: string; - cwd: string; - provider: string; - model: string; - scope_summary: string; - tools: Array; - /** - * Model context window in tokens. Always filled by the Worker greeting. - */ - context_window: number; - /** - * Estimated current session context tokens at connect time. - */ - context_tokens: number; -}; +export type Greeting = { worker_name: string, cwd: string, provider: string, model: string, scope_summary: string, tools: Array, +/** + * Model context window in tokens. Always filled by the Worker greeting. + */ +context_window: number, +/** + * Estimated current session context tokens at connect time. + */ +context_tokens: number, }; -export type Alert = { - level: AlertLevel; - source: AlertSource; - message: string; - /** - * Milliseconds since the Unix epoch. - */ - timestamp_ms: number; -}; +export type Alert = { level: AlertLevel, source: AlertSource, message: string, +/** + * Milliseconds since the Unix epoch. + */ +timestamp_ms: number, }; -export type MemoryWorkerEvent = { - worker: string; - status: string; - run_id: string; - trigger: string; - reason: string; - /** - * Human-readable compact form for actionbar rendering. - */ - message: string; - /** - * Milliseconds since the Unix epoch. - */ - timestamp_ms: number; -}; +export type MemoryWorkerEvent = { worker: string, status: string, run_id: string, trigger: string, reason: string, +/** + * Human-readable compact form for actionbar rendering. + */ +message: string, +/** + * Milliseconds since the Unix epoch. + */ +timestamp_ms: number, }; -export type Segment = - | { "kind": "text"; content: string } - | { - "kind": "paste"; - id: number; - chars: number; - lines: number; - content: string; - } - | { "kind": "file_ref"; path: string } - | { "kind": "unknown" }; +export type Segment = { "kind": "text", content: string, } | { "kind": "paste", id: number, chars: number, lines: number, content: string, } | { "kind": "file_ref", path: string, } | { "kind": "unknown" }; -export type WorkerEvent = - | { "kind": "turn_ended"; worker_name: string } - | { "kind": "errored"; worker_name: string; message: string } - | { "kind": "shut_down"; worker_name: string } - | { - "kind": "scope_sub_delegated"; - /** - * Sub-delegating Worker (= the sender itself). - */ - parent_worker: string; - /** - * Name of the grandchild Worker. - */ - sub_worker: string; - /** - * Unix-socket path where the grandchild is reachable. - */ - sub_socket: string; - /** - * Scope delegated to the grandchild. - */ - scope: Array; - }; +export type WorkerEvent = { "kind": "turn_ended", worker_name: string, } | { "kind": "errored", worker_name: string, message: string, } | { "kind": "shut_down", worker_name: string, } | { "kind": "scope_sub_delegated", +/** + * Sub-delegating Worker (= the sender itself). + */ +parent_worker: string, +/** + * Name of the grandchild Worker. + */ +sub_worker: string, +/** + * Unix-socket path where the grandchild is reachable. + */ +sub_socket: string, +/** + * Scope delegated to the grandchild. + */ +scope: Array, }; -export type Method = - | { "method": "run"; "params": { input: Array } } - | { "method": "notify"; "params": { message: string; auto_run?: boolean } } - | { "method": "worker_event"; "params": WorkerEvent } - | { "method": "resume" } - | { "method": "cancel" } - | { "method": "pause" } - | { "method": "compact" } - | { "method": "list_rewind_targets" } - | { - "method": "rewind_to"; - "params": { target: RewindTargetId; expected_head_entries: number }; - } - | { "method": "shutdown" } - | { - "method": "list_completions"; - "params": { kind: CompletionKind; prefix: string }; - } - | { "method": "list_workers" } - | { "method": "restore_worker"; "params": { name: string } } - | { "method": "register_peer"; "params": { name: string } }; +export type SubscriptionRequestId = string; -export type Event = - | { "event": "user_message"; "data": { segments: Array } } - | { "event": "system_item"; "data": { item: unknown } } - | { "event": "invoke_start"; "data": { kind: InvokeKind } } - | { "event": "turn_start"; "data": { turn: number } } - | { "event": "turn_end"; "data": { turn: number; result: TurnResult } } - | { "event": "llm_call_start"; "data": { llm_call: number } } - | { "event": "llm_call_end"; "data": { llm_call: number } } - | { - "event": "llm_retry"; - "data": { - llm_call: number; - /** - * The attempt that just failed. 1 origin. - */ - failed_attempt: number; - max_attempts: number; - wait_ms: number; - elapsed_ms: number; - status?: number | null; - error: string; - }; - } - | { - "event": "llm_continuation"; - "data": { - llm_call: number; - attempt: number; - max_attempts: number; - reason: string; - }; - } - | { "event": "text_delta"; "data": { text: string } } - | { "event": "text_done"; "data": { text: string } } - | { "event": "thinking_start" } - | { "event": "thinking_delta"; "data": { text: string } } - | { "event": "thinking_done"; "data": { text: string } } - | { "event": "tool_call_start"; "data": { id: string; name: string } } - | { "event": "tool_call_args_delta"; "data": { id: string; json: string } } - | { - "event": "tool_call_done"; - "data": { id: string; name: string; arguments: string }; - } - | { - "event": "tool_result"; - "data": { - id: string; - /** - * Short human-readable summary. Always present; used by clients - * that only want a 1-line rendering (e.g. collapsed views). - */ - summary: string; - /** - * Full tool output. Absent when the tool chose to return - * summary-only, or when the result was pruned. - */ - output?: string | null; - is_error: boolean; - }; - } - | { - "event": "usage"; - "data": { - input_tokens: number | null; - output_tokens: number | null; - cache_read_input_tokens?: number | null; - }; - } - | { "event": "run_end"; "data": { result: RunResult } } - | { "event": "error"; "data": { code: ErrorCode; message: string } } - | { - "event": "snapshot"; - "data": { - entries: Array; - greeting: Greeting; - status: WorkerStatus; - /** - * Unfinished model output that has already streamed in the current - * run but is not yet represented by committed snapshot entries. - */ - in_flight?: InFlightSnapshot; - }; - } - | { "event": "segment_rotated"; "data": { entry: unknown } } - | { "event": "status"; "data": { status: WorkerStatus } } - | { - "event": "completions"; - "data": { kind: CompletionKind; entries: Array }; - } - | { - "event": "rewind_targets"; - "data": { head_entries: number; targets: Array }; - } - | { - "event": "rewind_applied"; - "data": { - entries: Array; - input: Array; - summary: RewindSummary; - }; - } - | { "event": "workers_listed"; "data": { workers: unknown } } - | { "event": "worker_restored"; "data": { result: unknown } } - | { "event": "peer_registered"; "data": { result: unknown } } - | { "event": "alert"; "data": Alert } - | { "event": "memory_worker"; "data": MemoryWorkerEvent } - | { "event": "compact_start" } - | { "event": "compact_done"; "data": { new_segment_id: string } } - | { "event": "compact_failed"; "data": { error: string } } - | { "event": "shutdown" }; +export type SubscriptionId = string; + +export type SubscriptionWorkerId = string; + +export type SubscriptionWorkdirId = string; + +export type SubscriptionWorkerIds = Array; + +export type SubscriptionWorkerState = "idle" | "running" | "paused" | "stopped" | "cancelled"; + +export type EventSubscriptionSelector = { "topic": "runtime_workers" } | { "topic": "worker_lifecycle", worker_ids: SubscriptionWorkerIds, } | { "topic": "worker_protocol", worker_id: SubscriptionWorkerId, } | { "topic": "workspace_workers" } | { "topic": "workspace_workdirs" }; + +export type SubscriptionWorker = { worker_id: SubscriptionWorkerId, +/** + * Producer-owned monotonic revision for this Worker subject. + */ +subject_revision: number, state: SubscriptionWorkerState, workspace_id?: string | null, display_name?: string | null, profile?: string | null, working_directory_id?: SubscriptionWorkdirId | null, }; + +export type SubscriptionWorkdir = { working_directory_id: SubscriptionWorkdirId, repository_id: string, state: string, primary_worker_id?: SubscriptionWorkerId | null, }; + +export type SubscriptionSnapshot = { "topic": "workers", "data": { workers: Array, } } | { "topic": "worker_protocol", "data": { worker_id: SubscriptionWorkerId, events: Array, } } | { "topic": "workspace_workdirs", "data": { workdirs: Array, } }; + +export type SubscriptionEventPayload = { "event": "worker_upserted", "data": { worker: SubscriptionWorker, } } | { "event": "worker_removed", "data": { worker_id: SubscriptionWorkerId, } } | { "event": "worker_protocol", "data": { worker_id: SubscriptionWorkerId, event: Event, } } | { "event": "workdir_upserted", "data": { workdir: SubscriptionWorkdir, } } | { "event": "workdir_removed", "data": { working_directory_id: SubscriptionWorkdirId, } }; + +export type SubscriptionRejectionCode = "invalid_request" | "unsupported_protocol_version" | "unsupported_selector" | "unauthorized" | "resource_not_found" | "capacity_exceeded" | "internal"; + +export type SubscriptionTerminationCode = "lagged" | "resource_gone" | "unauthorized" | "server_shutdown"; + +export type SubscriptionRequest = { "method": "subscribe_events", "params": { request_id: SubscriptionRequestId, selector: EventSubscriptionSelector, } } | { "method": "unsubscribe_events", "params": { request_id: SubscriptionRequestId, subscription_id: SubscriptionId, } }; + +export type SubscriptionResponse = { "result": "subscribed", "payload": { request_id: SubscriptionRequestId, subscription_id: SubscriptionId, selector: EventSubscriptionSelector, snapshot_revision: number, snapshot: SubscriptionSnapshot, } } | { "result": "unsubscribed", "payload": { request_id: SubscriptionRequestId, subscription_id: SubscriptionId, } } | { "result": "subscription_rejected", "payload": { request_id: SubscriptionRequestId, subscription_id?: SubscriptionId | null, code: SubscriptionRejectionCode, message: string, } }; + +export type SubscriptionEvent = { "event": "event", "data": { subscription_id: SubscriptionId, subject_revision: number, payload: SubscriptionEventPayload, } } | { "event": "subscription_closed", "data": { subscription_id: SubscriptionId, code: SubscriptionTerminationCode, message: string, } }; + +export type SubscriptionFramePayload = { "frame": "request", "message": SubscriptionRequest } | { "frame": "response", "message": SubscriptionResponse } | { "frame": "event", "message": SubscriptionEvent }; + +export type SubscriptionFrame = { protocol_version: number, } & ({ "frame": "request", "message": SubscriptionRequest } | { "frame": "response", "message": SubscriptionResponse } | { "frame": "event", "message": SubscriptionEvent }); + +export type Method = { "method": "run", "params": { input: Array, } } | { "method": "notify", "params": { message: string, auto_run?: boolean, } } | { "method": "worker_event", "params": WorkerEvent } | { "method": "resume" } | { "method": "cancel" } | { "method": "pause" } | { "method": "compact" } | { "method": "list_rewind_targets" } | { "method": "rewind_to", "params": { target: RewindTargetId, expected_head_entries: number, } } | { "method": "shutdown" } | { "method": "list_completions", "params": { kind: CompletionKind, prefix: string, } } | { "method": "list_workers" } | { "method": "restore_worker", "params": { name: string, } } | { "method": "register_peer", "params": { name: string, } }; + +export type Event = { "event": "user_message", "data": { segments: Array, } } | { "event": "system_item", "data": { item: unknown, } } | { "event": "invoke_start", "data": { kind: InvokeKind, } } | { "event": "turn_start", "data": { turn: number, } } | { "event": "turn_end", "data": { turn: number, result: TurnResult, } } | { "event": "llm_call_start", "data": { llm_call: number, } } | { "event": "llm_call_end", "data": { llm_call: number, } } | { "event": "llm_retry", "data": { llm_call: number, +/** + * The attempt that just failed. 1 origin. + */ +failed_attempt: number, max_attempts: number, wait_ms: number, elapsed_ms: number, status?: number | null, error: string, } } | { "event": "llm_continuation", "data": { llm_call: number, attempt: number, max_attempts: number, reason: string, } } | { "event": "text_delta", "data": { text: string, } } | { "event": "text_done", "data": { text: string, } } | { "event": "thinking_start" } | { "event": "thinking_delta", "data": { text: string, } } | { "event": "thinking_done", "data": { text: string, } } | { "event": "tool_call_start", "data": { id: string, name: string, } } | { "event": "tool_call_args_delta", "data": { id: string, json: string, } } | { "event": "tool_call_done", "data": { id: string, name: string, arguments: string, } } | { "event": "tool_result", "data": { id: string, +/** + * Short human-readable summary. Always present; used by clients + * that only want a 1-line rendering (e.g. collapsed views). + */ +summary: string, +/** + * Full tool output. Absent when the tool chose to return + * summary-only, or when the result was pruned. + */ +output?: string | null, is_error: boolean, } } | { "event": "usage", "data": { input_tokens: number | null, output_tokens: number | null, cache_read_input_tokens?: number | null, } } | { "event": "run_end", "data": { result: RunResult, } } | { "event": "error", "data": { code: ErrorCode, message: string, } } | { "event": "snapshot", "data": { entries: Array, greeting: Greeting, status: WorkerStatus, +/** + * Unfinished model output that has already streamed in the current + * run but is not yet represented by committed snapshot entries. + */ +in_flight?: InFlightSnapshot, } } | { "event": "segment_rotated", "data": { entry: unknown, } } | { "event": "status", "data": { status: WorkerStatus, } } | { "event": "completions", "data": { kind: CompletionKind, entries: Array, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array, } } | { "event": "rewind_applied", "data": { entries: Array, input: Array, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compact_start" } | { "event": "compact_done", "data": { new_segment_id: string, } } | { "event": "compact_failed", "data": { error: string, } } | { "event": "shutdown" };