diff --git a/src/common/active_session_snapshot.rs b/src/common/active_session_snapshot.rs new file mode 100644 index 0000000..a39bf9e --- /dev/null +++ b/src/common/active_session_snapshot.rs @@ -0,0 +1,525 @@ +use serde::{Deserialize, Serialize}; + +use super::SessionId; + +/// Initial schema version for active-session restore snapshots. +pub const ACTIVE_RESTORE_SNAPSHOT_SCHEMA_VERSION_V1: u32 = 1; + +/// Initial reducer version for active-session restore snapshots. +pub const ACTIVE_RESTORE_REDUCER_VERSION_V1: u32 = 1; + +/// Storage layouts a client can read. +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub enum ActiveSessionSnapshotStorageLayoutKind { + MonolithicV1, +} + +/// Compression codecs supported for immutable snapshot objects. +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub enum ActiveSessionSnapshotCompression { + Zstd, +} + +/// Snapshot capabilities advertised by a client. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct ActiveSessionSnapshotCapabilities { + pub snapshot_schema_versions: Vec, + pub reducer_versions: Vec, + pub storage_layouts: Vec, + pub compression_codecs: Vec, +} + +impl ActiveSessionSnapshotCapabilities { + pub fn supports(&self, protocol: &NegotiatedActiveSessionSnapshotProtocol) -> bool { + self.snapshot_schema_versions + .contains(&protocol.snapshot_schema_version) + && self.reducer_versions.contains(&protocol.reducer_version) + && self.storage_layouts.contains(&protocol.storage_layout) + && self + .compression_codecs + .contains(&protocol.compression_codec) + } +} + +/// Snapshot protocol values positively selected by the server. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct NegotiatedActiveSessionSnapshotProtocol { + pub snapshot_schema_version: u32, + pub reducer_version: u32, + pub storage_layout: ActiveSessionSnapshotStorageLayoutKind, + pub compression_codec: ActiveSessionSnapshotCompression, +} + +/// Exact session and active execution represented by a snapshot. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct ActiveSessionSnapshotIdentity { + pub session_id: SessionId, + pub conversation_id: String, + /// Opaque server-issued execution epoch. Clients must not derive this value. + pub execution_id: String, + /// Stable conversation-run correlation key across execution handoffs. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_id: Option, +} + +/// A protocol selection bound to the exact execution being restored. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct NegotiatedActiveSessionSnapshot { + pub protocol: NegotiatedActiveSessionSnapshotProtocol, + pub identity: ActiveSessionSnapshotIdentity, +} + +/// The downloaded logical snapshot payload. +/// +/// Storage metadata is deliberately outside this payload so its hash does not +/// recursively include itself. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct ActiveRestoreSnapshot { + pub snapshot_id: String, + pub identity: ActiveSessionSnapshotIdentity, + pub snapshot_schema_version: u32, + pub reducer_version: u32, + pub through_event_no: u64, + pub captured_at_unix_ms: u64, + pub ordered_message_ids: Vec, + pub terminal_state: Vec, + pub conversation_data: Vec, + pub additional_reducer_state: Vec, +} + +/// An immutable object containing a compressed active-session snapshot. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct ImmutableActiveSessionSnapshotObject { + /// Opaque storage-system identifier. This is not a download URL. + pub object_id: String, + /// Opaque immutable object generation. + pub object_generation: String, + /// Lowercase hexadecimal SHA-256 of the compressed object bytes. + pub sha256: String, + pub compressed_size_bytes: u64, + pub uncompressed_size_bytes: u64, + pub compression_codec: ActiveSessionSnapshotCompression, +} + +/// Layout-specific storage descriptor for an active-session snapshot. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub enum ActiveSessionSnapshotStorage { + MonolithicV1 { + object: ImmutableActiveSessionSnapshotObject, + }, +} + +impl ActiveSessionSnapshotStorage { + pub fn kind(&self) -> ActiveSessionSnapshotStorageLayoutKind { + match self { + Self::MonolithicV1 { .. } => ActiveSessionSnapshotStorageLayoutKind::MonolithicV1, + } + } + + pub fn compression_codec(&self) -> ActiveSessionSnapshotCompression { + match self { + Self::MonolithicV1 { object } => object.compression_codec, + } + } +} + +/// Immutable metadata needed to select and download one snapshot generation. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct ActiveSessionSnapshotDescriptor { + pub snapshot_id: String, + pub identity: ActiveSessionSnapshotIdentity, + pub snapshot_schema_version: u32, + pub reducer_version: u32, + pub through_event_no: u64, + pub captured_at_unix_ms: u64, + pub storage: ActiveSessionSnapshotStorage, +} + +impl ActiveSessionSnapshotDescriptor { + pub fn negotiated_protocol(&self) -> NegotiatedActiveSessionSnapshotProtocol { + NegotiatedActiveSessionSnapshotProtocol { + snapshot_schema_version: self.snapshot_schema_version, + reducer_version: self.reducer_version, + storage_layout: self.storage.kind(), + compression_codec: self.storage.compression_codec(), + } + } +} + +/// A storage-validated snapshot that may be submitted for publication. +#[derive(Clone, Deserialize, Serialize, PartialEq, Eq)] +pub struct PreparedActiveSessionSnapshotReceipt { + pub receipt: String, + pub descriptor: ActiveSessionSnapshotDescriptor, +} + +impl std::fmt::Debug for PreparedActiveSessionSnapshotReceipt { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PreparedActiveSessionSnapshotReceipt") + .field("receipt", &"***") + .field("descriptor", &self.descriptor) + .finish() + } +} + +/// Result of preparing or publishing a snapshot generation. +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub enum ActiveSessionSnapshotPublicationStatus { + Prepared, + Committed, + Idempotent, + Stale, + NotContiguousYet, + Conflict, + TooLarge, + UnsupportedVersion, + InvalidObject, +} + +/// Acknowledgement for a snapshot publication attempt. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct ActiveSessionSnapshotPublicationAck { + pub snapshot_id: String, + pub status: ActiveSessionSnapshotPublicationStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub highest_contiguous_event_no: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retained_event_floor: Option, +} + +/// Cursor supplied by a viewer reconnecting after snapshot restoration. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct ActiveSessionSnapshotResumeCursor { + pub snapshot_id: String, + pub identity: ActiveSessionSnapshotIdentity, + pub snapshot_schema_version: u32, + pub reducer_version: u32, + pub storage_layout: ActiveSessionSnapshotStorageLayoutKind, + pub compression_codec: ActiveSessionSnapshotCompression, + pub last_contiguous_event_no: u64, +} + +impl ActiveSessionSnapshotResumeCursor { + pub fn protocol(&self) -> NegotiatedActiveSessionSnapshotProtocol { + NegotiatedActiveSessionSnapshotProtocol { + snapshot_schema_version: self.snapshot_schema_version, + reducer_version: self.reducer_version, + storage_layout: self.storage_layout, + compression_codec: self.compression_codec, + } + } +} + +/// Restore instructions returned after a viewer joins or reconnects. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub enum ActiveSessionSnapshotRestore { + Bootstrap { + negotiated: NegotiatedActiveSessionSnapshot, + snapshot: Box, + resume_from_event_no: u64, + catch_up_through_event_no: u64, + retained_event_floor: u64, + }, + Resume { + negotiated: NegotiatedActiveSessionSnapshot, + snapshot_id: String, + resume_from_event_no: u64, + catch_up_through_event_no: u64, + retained_event_floor: u64, + }, +} + +/// Why a viewer must discard transient state and request a fresh bootstrap. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub enum ActiveSessionSnapshotResyncReason { + SessionChanged, + ConversationChanged, + ExecutionChanged, + SnapshotSchemaMismatch { + expected: u32, + received: u32, + }, + ReducerVersionMismatch { + expected: u32, + received: u32, + }, + StorageLayoutMismatch, + SnapshotIdentityMismatch, + CursorAhead { + highest_contiguous_event_no: u64, + }, + CursorBelowRetainedFloor { + retained_event_floor: u64, + }, + SnapshotMissing, + EventMissing { + event_no: u64, + }, + SnapshotHashMismatch, + SnapshotSizeMismatch, + SnapshotDecompressionFailed, + SnapshotDecodeFailed, + ReplayGap { + expected_event_no: u64, + next_available_event_no: u64, + }, + ConflictingDuplicate { + event_no: u64, + }, + StorageUnavailable, + BufferLimitExceeded, + BootstrapTimedOut, +} + +/// Fail-closed validation error for negotiated snapshot restore instructions. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ActiveSessionSnapshotValidationError { + UnsupportedProtocol, + ProtocolMismatch, + IdentityMismatch, + SnapshotIdMismatch, + SnapshotMetadataMismatch, + InvalidResumeCursor, +} + +pub fn validate_active_restore_snapshot( + descriptor: &ActiveSessionSnapshotDescriptor, + snapshot: &ActiveRestoreSnapshot, +) -> Result<(), ActiveSessionSnapshotValidationError> { + if descriptor.snapshot_id != snapshot.snapshot_id { + return Err(ActiveSessionSnapshotValidationError::SnapshotIdMismatch); + } + if descriptor.identity != snapshot.identity { + return Err(ActiveSessionSnapshotValidationError::IdentityMismatch); + } + if descriptor.snapshot_schema_version != snapshot.snapshot_schema_version + || descriptor.reducer_version != snapshot.reducer_version + || descriptor.through_event_no != snapshot.through_event_no + || descriptor.captured_at_unix_ms != snapshot.captured_at_unix_ms + { + return Err(ActiveSessionSnapshotValidationError::SnapshotMetadataMismatch); + } + Ok(()) +} + +pub fn validate_active_session_snapshot_restore( + capabilities: &ActiveSessionSnapshotCapabilities, + restore: &ActiveSessionSnapshotRestore, +) -> Result<(), ActiveSessionSnapshotValidationError> { + match restore { + ActiveSessionSnapshotRestore::Bootstrap { + negotiated, + snapshot, + resume_from_event_no, + catch_up_through_event_no, + retained_event_floor, + } => { + if !capabilities.supports(&negotiated.protocol) { + return Err(ActiveSessionSnapshotValidationError::UnsupportedProtocol); + } + if snapshot.negotiated_protocol() != negotiated.protocol { + return Err(ActiveSessionSnapshotValidationError::ProtocolMismatch); + } + if snapshot.identity != negotiated.identity { + return Err(ActiveSessionSnapshotValidationError::IdentityMismatch); + } + if snapshot.through_event_no.checked_add(1) != Some(*resume_from_event_no) + || *resume_from_event_no > catch_up_through_event_no.saturating_add(1) + || *retained_event_floor > *resume_from_event_no + { + return Err(ActiveSessionSnapshotValidationError::InvalidResumeCursor); + } + } + ActiveSessionSnapshotRestore::Resume { + negotiated, + resume_from_event_no, + catch_up_through_event_no, + retained_event_floor, + .. + } => { + if !capabilities.supports(&negotiated.protocol) { + return Err(ActiveSessionSnapshotValidationError::UnsupportedProtocol); + } + if *resume_from_event_no > catch_up_through_event_no.saturating_add(1) + || *retained_event_floor > *resume_from_event_no + { + return Err(ActiveSessionSnapshotValidationError::InvalidResumeCursor); + } + } + } + + Ok(()) +} + +pub fn validate_active_session_snapshot_resume( + cursor: &ActiveSessionSnapshotResumeCursor, + restore: &ActiveSessionSnapshotRestore, +) -> Result<(), ActiveSessionSnapshotValidationError> { + let ActiveSessionSnapshotRestore::Resume { + negotiated, + snapshot_id, + resume_from_event_no, + .. + } = restore + else { + return Err(ActiveSessionSnapshotValidationError::InvalidResumeCursor); + }; + + if snapshot_id != &cursor.snapshot_id { + return Err(ActiveSessionSnapshotValidationError::SnapshotIdMismatch); + } + if negotiated.identity != cursor.identity { + return Err(ActiveSessionSnapshotValidationError::IdentityMismatch); + } + if negotiated.protocol != cursor.protocol() { + return Err(ActiveSessionSnapshotValidationError::ProtocolMismatch); + } + if cursor.last_contiguous_event_no.checked_add(1) != Some(*resume_from_event_no) { + return Err(ActiveSessionSnapshotValidationError::InvalidResumeCursor); + } + Ok(()) +} +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use super::*; + + fn identity() -> ActiveSessionSnapshotIdentity { + ActiveSessionSnapshotIdentity { + session_id: SessionId::from_str("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa").unwrap(), + conversation_id: "conversation".into(), + execution_id: "314159".into(), + run_id: Some("run".into()), + } + } + + fn capabilities() -> ActiveSessionSnapshotCapabilities { + ActiveSessionSnapshotCapabilities { + snapshot_schema_versions: vec![ACTIVE_RESTORE_SNAPSHOT_SCHEMA_VERSION_V1], + reducer_versions: vec![ACTIVE_RESTORE_REDUCER_VERSION_V1], + storage_layouts: vec![ActiveSessionSnapshotStorageLayoutKind::MonolithicV1], + compression_codecs: vec![ActiveSessionSnapshotCompression::Zstd], + } + } + + fn descriptor() -> ActiveSessionSnapshotDescriptor { + ActiveSessionSnapshotDescriptor { + snapshot_id: "snapshot".into(), + identity: identity(), + snapshot_schema_version: ACTIVE_RESTORE_SNAPSHOT_SCHEMA_VERSION_V1, + reducer_version: ACTIVE_RESTORE_REDUCER_VERSION_V1, + through_event_no: 41, + captured_at_unix_ms: 1_700_000_000_000, + storage: ActiveSessionSnapshotStorage::MonolithicV1 { + object: ImmutableActiveSessionSnapshotObject { + object_id: "object".into(), + object_generation: "7".into(), + sha256: "00".repeat(32), + compressed_size_bytes: 512, + uncompressed_size_bytes: 1024, + compression_codec: ActiveSessionSnapshotCompression::Zstd, + }, + }, + } + } + + #[test] + fn monolithic_bootstrap_requires_exact_protocol_and_cursor() { + let descriptor = descriptor(); + let restore = ActiveSessionSnapshotRestore::Bootstrap { + negotiated: NegotiatedActiveSessionSnapshot { + protocol: descriptor.negotiated_protocol(), + identity: identity(), + }, + snapshot: Box::new(descriptor), + resume_from_event_no: 42, + catch_up_through_event_no: 45, + retained_event_floor: 42, + }; + + assert!(validate_active_session_snapshot_restore(&capabilities(), &restore).is_ok()); + } + + #[test] + fn bootstrap_rejects_an_identity_mismatch() { + let descriptor = descriptor(); + let mut negotiated_identity = identity(); + negotiated_identity.execution_id = "different-execution".into(); + let restore = ActiveSessionSnapshotRestore::Bootstrap { + negotiated: NegotiatedActiveSessionSnapshot { + protocol: descriptor.negotiated_protocol(), + identity: negotiated_identity, + }, + snapshot: Box::new(descriptor), + resume_from_event_no: 42, + catch_up_through_event_no: 45, + retained_event_floor: 42, + }; + + assert_eq!( + validate_active_session_snapshot_restore(&capabilities(), &restore), + Err(ActiveSessionSnapshotValidationError::IdentityMismatch) + ); + } + + #[test] + fn downloaded_snapshot_must_match_its_descriptor() { + let descriptor = descriptor(); + let mut snapshot = ActiveRestoreSnapshot { + snapshot_id: descriptor.snapshot_id.clone(), + identity: descriptor.identity.clone(), + snapshot_schema_version: descriptor.snapshot_schema_version, + reducer_version: descriptor.reducer_version, + through_event_no: descriptor.through_event_no, + captured_at_unix_ms: descriptor.captured_at_unix_ms, + ordered_message_ids: vec!["message-1".into()], + terminal_state: vec![1], + conversation_data: vec![2], + additional_reducer_state: vec![3], + }; + + assert!(validate_active_restore_snapshot(&descriptor, &snapshot).is_ok()); + snapshot.through_event_no += 1; + assert_eq!( + validate_active_restore_snapshot(&descriptor, &snapshot), + Err(ActiveSessionSnapshotValidationError::SnapshotMetadataMismatch) + ); + } + + #[test] + fn resume_requires_the_exact_previous_cursor() { + let descriptor = descriptor(); + let cursor = ActiveSessionSnapshotResumeCursor { + snapshot_id: descriptor.snapshot_id.clone(), + identity: descriptor.identity.clone(), + snapshot_schema_version: descriptor.snapshot_schema_version, + reducer_version: descriptor.reducer_version, + storage_layout: descriptor.storage.kind(), + compression_codec: descriptor.storage.compression_codec(), + last_contiguous_event_no: 44, + }; + let restore = ActiveSessionSnapshotRestore::Resume { + negotiated: NegotiatedActiveSessionSnapshot { + protocol: descriptor.negotiated_protocol(), + identity: descriptor.identity, + }, + snapshot_id: descriptor.snapshot_id, + resume_from_event_no: 45, + catch_up_through_event_no: 47, + retained_event_floor: 42, + }; + + assert!(validate_active_session_snapshot_resume(&cursor, &restore).is_ok()); + } + + #[test] + fn unknown_storage_layout_fails_closed() { + assert!( + serde_json::from_str::( + r#"{"ChunkedV2":{"manifest":{"object_id":"manifest"}}}"# + ) + .is_err() + ); + } +} diff --git a/src/common/feature_support.rs b/src/common/feature_support.rs index b9e810c..f1eebca 100644 --- a/src/common/feature_support.rs +++ b/src/common/feature_support.rs @@ -1,3 +1,4 @@ +use super::ActiveSessionSnapshotCapabilities; use serde::{Deserialize, Serialize}; /// Client feature support declaration. @@ -14,4 +15,7 @@ pub struct FeatureSupport { /// Whether the client supports the "Full" role ACL. #[serde(default)] pub supports_full_role_for_real: bool, + /// Active-session snapshot protocol versions this client can use. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub active_session_snapshot: Option, } diff --git a/src/common/mod.rs b/src/common/mod.rs index e4d755f..4f33f78 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -1,4 +1,5 @@ //! Common types used by both sharer and viewer. +mod active_session_snapshot; mod agent_prompt; mod command_execution; @@ -19,6 +20,7 @@ mod ui_state; mod user; mod write_to_pty; +pub use active_session_snapshot::*; pub use agent_prompt::*; pub use command_execution::*; pub use control_action::*; diff --git a/src/sharer.rs b/src/sharer.rs index 93898ac..8db8fb1 100644 --- a/src/sharer.rs +++ b/src/sharer.rs @@ -12,14 +12,16 @@ //! since old clients may not specify new fields expected by the server. use crate::common::{ - ActivePrompt, ActivePromptUpdate, AgentPromptFailureReason, AgentPromptRequest, - AgentPromptRequestId, BlockId, BufferId, CommandExecutionFailureReason, - CommandExecutionRequestId, ControlAction, ControlActionFailureReason, ControlActionRequestId, - FeatureSupport, InputOperationId, InputReplicaId, InputUpdate, InputUpdateFailureReason, - OrderedTerminalEvent, ParticipantId, ParticipantList, ParticipantPresenceUpdate, Role, - RoleRequestId, RoleRequestResponse, Selection, SelectionUpdate, SessionId, SessionSecret, - TelemetryContext, UniversalDeveloperInputContext, UniversalDeveloperInputContextUpdate, UserID, - WindowSize, WriteToPtyFailureReason, WriteToPtyRequestId, + ActivePrompt, ActivePromptUpdate, ActiveSessionSnapshotPublicationAck, + AgentPromptFailureReason, AgentPromptRequest, AgentPromptRequestId, BlockId, BufferId, + CommandExecutionFailureReason, CommandExecutionRequestId, ControlAction, + ControlActionFailureReason, ControlActionRequestId, FeatureSupport, InputOperationId, + InputReplicaId, InputUpdate, InputUpdateFailureReason, NegotiatedActiveSessionSnapshotProtocol, + OrderedTerminalEvent, ParticipantId, ParticipantList, ParticipantPresenceUpdate, + PreparedActiveSessionSnapshotReceipt, Role, RoleRequestId, RoleRequestResponse, Selection, + SelectionUpdate, SessionId, SessionSecret, TelemetryContext, UniversalDeveloperInputContext, + UniversalDeveloperInputContextUpdate, UserID, WindowSize, WriteToPtyFailureReason, + WriteToPtyRequestId, }; use super::common::Scrollback; @@ -368,6 +370,9 @@ pub enum DownstreamMessage { sharer_id: ParticipantId, /// The Firebase UID assigned to the sharer. sharer_firebase_uid: String, + /// Snapshot protocol selected from the sharer's advertised capabilities. + #[serde(default, skip_serializing_if = "Option::is_none")] + active_session_snapshot_protocol: Option, }, /// The server denied the initialization request. No further messages will be processed. @@ -385,6 +390,9 @@ pub enum DownstreamMessage { /// The sharer can use this to update the server with any newer events created while disconnected. last_received_event_no: Option, participant_list: ParticipantList, + /// Snapshot protocol selected from the sharer's advertised capabilities. + #[serde(default, skip_serializing_if = "Option::is_none")] + active_session_snapshot_protocol: Option, }, /// The server denied the reconnection request. No further messages will be processed. @@ -394,6 +402,9 @@ pub enum DownstreamMessage { /// and the sharer can safely remove them from memory. EventsProcessedAck { latest_processed_event_no: usize }, + /// The server acknowledged an out-of-band snapshot publication attempt. + ActiveSessionSnapshotPublicationAck(ActiveSessionSnapshotPublicationAck), + /// Sent when the list of participants in the shared session changes. ParticipantListUpdated(ParticipantList), @@ -485,6 +496,18 @@ pub enum DownstreamMessage { } impl DownstreamMessage { + pub fn requires_active_session_snapshot_support(&self) -> bool { + matches!( + self, + Self::SessionInitialized { + active_session_snapshot_protocol: Some(_), + .. + } | Self::SessionReconnected { + active_session_snapshot_protocol: Some(_), + .. + } | Self::ActiveSessionSnapshotPublicationAck(_) + ) + } pub fn from_json(json: &str) -> serde_json::Result { serde_json::from_str(json) } @@ -530,6 +553,11 @@ pub enum UpstreamMessage { /// Sent when there is any ordered terminal event. OrderedTerminalEvent(OrderedTerminalEvent), + /// Submits a storage-validated snapshot receipt for publication. + PublishActiveSessionSnapshot { + receipt: PreparedActiveSessionSnapshotReceipt, + }, + /// Sent to reconnect to the server after disconnection. Reconnect(ReconnectPayload), diff --git a/src/viewer.rs b/src/viewer.rs index 5295378..0cc2149 100644 --- a/src/viewer.rs +++ b/src/viewer.rs @@ -15,15 +15,17 @@ use crate::{ common::{ - ActivePrompt, ActivePromptUpdate, AgentAttachment, AgentPromptFailureReason, - AgentPromptRequest, AgentPromptRequestId, BlockId, BufferId, CommandExecutionFailureReason, - CommandExecutionRequestId, ControlAction, ControlActionFailureReason, FeatureSupport, - InputOperationId, InputReplicaId, InputUpdate, InputUpdateFailureReason, - LinkAccessLevelUpdateResponse, OrderedTerminalEvent, ParticipantId, ParticipantList, - ParticipantPresenceUpdate, Role, RoleRequestId, RoleRequestResponse, Scrollback, - SelectionUpdate, TeamAccessLevelUpdateResponse, TeamAclData, TelemetryContext, - UniversalDeveloperInputContext, UniversalDeveloperInputContextUpdate, UserID, WindowSize, - WriteToPtyFailureReason, WriteToPtyRequestId, + ActivePrompt, ActivePromptUpdate, ActiveSessionSnapshotRestore, + ActiveSessionSnapshotResumeCursor, ActiveSessionSnapshotResyncReason, AgentAttachment, + AgentPromptFailureReason, AgentPromptRequest, AgentPromptRequestId, BlockId, BufferId, + CommandExecutionFailureReason, CommandExecutionRequestId, ControlAction, + ControlActionFailureReason, FeatureSupport, InputOperationId, InputReplicaId, InputUpdate, + InputUpdateFailureReason, LinkAccessLevelUpdateResponse, OrderedTerminalEvent, + ParticipantId, ParticipantList, ParticipantPresenceUpdate, Role, RoleRequestId, + RoleRequestResponse, Scrollback, SelectionUpdate, TeamAccessLevelUpdateResponse, + TeamAclData, TelemetryContext, UniversalDeveloperInputContext, + UniversalDeveloperInputContextUpdate, UserID, WindowSize, WriteToPtyFailureReason, + WriteToPtyRequestId, }, sharer::{self, LegacySessionSourceType, SessionSourceType}, }; @@ -100,6 +102,10 @@ pub struct InitPayload { /// Client feature support declaration. #[serde(default)] pub feature_support: FeatureSupport, + + /// Snapshot identity and last contiguous event applied by a reconnecting viewer. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub active_session_snapshot_cursor: Option, } /// The possible messages sent from server to client (viewer). @@ -149,11 +155,18 @@ pub enum DownstreamMessage { /// off the source-type variant kind. #[serde(default)] source_task_id: Option, + + /// Negotiated snapshot bootstrap instructions. Absent for legacy replay. + #[serde(default, skip_serializing_if = "Option::is_none")] + active_session_snapshot_restore: Option>, }, /// The server sends this message when the session was successfully rejoined. RejoinedSuccessfully { participant_list: Box, + /// Negotiated resume or fresh bootstrap instructions. + #[serde(default, skip_serializing_if = "Option::is_none")] + active_session_snapshot_restore: Option>, }, /// Sent when the viewer fails to join the shared session. @@ -174,6 +187,11 @@ pub enum DownstreamMessage { /// These messages are only sent _after_ [`DownstreamMessage::JoinedSuccessfully`]. OrderedTerminalEvent(OrderedTerminalEvent), + /// The viewer must discard transient restore state and request a fresh bootstrap. + ActiveSessionSnapshotResyncRequired { + reason: ActiveSessionSnapshotResyncReason, + }, + /// Sent when the list of participants in the shared session changes. ParticipantListUpdated(ParticipantList), @@ -269,6 +287,18 @@ pub enum DownstreamMessage { } impl DownstreamMessage { + pub fn requires_active_session_snapshot_support(&self) -> bool { + matches!( + self, + Self::JoinedSuccessfully { + active_session_snapshot_restore: Some(_), + .. + } | Self::RejoinedSuccessfully { + active_session_snapshot_restore: Some(_), + .. + } | Self::ActiveSessionSnapshotResyncRequired { .. } + ) + } pub fn from_json(json: &str) -> serde_json::Result { serde_json::from_str(json) } @@ -285,9 +315,9 @@ impl DownstreamMessage { Self::JoinedSuccessfully { participant_list, .. } => participant_list.downgrade_full_roles(), - Self::RejoinedSuccessfully { participant_list } => { - participant_list.downgrade_full_roles() - } + Self::RejoinedSuccessfully { + participant_list, .. + } => participant_list.downgrade_full_roles(), Self::ParticipantListUpdated(list) => list.downgrade_full_roles(), Self::ParticipantRoleChanged { role, .. } => role.downgrade_full(), Self::RoleRequestResponse(RoleRequestResponse::Approved { new_role }) => { @@ -319,6 +349,8 @@ impl DownstreamMessage { } /// The possible messages sent from client (viewer) to server. +// Boxing `Initialize` would be wire-compatible but would churn every legacy call site. +#[allow(clippy::large_enum_variant)] #[derive(Debug, Serialize, Deserialize)] pub enum UpstreamMessage { /// The client sends this message to join the shared session. diff --git a/tests/active_session_snapshot_json_compatibility.rs b/tests/active_session_snapshot_json_compatibility.rs new file mode 100644 index 0000000..c35e6fe --- /dev/null +++ b/tests/active_session_snapshot_json_compatibility.rs @@ -0,0 +1,192 @@ +use session_sharing_protocol::{ + common::{ + ACTIVE_RESTORE_REDUCER_VERSION_V1, ACTIVE_RESTORE_SNAPSHOT_SCHEMA_VERSION_V1, + ActiveSessionSnapshotCapabilities, ActiveSessionSnapshotCompression, + ActiveSessionSnapshotPublicationAck, ActiveSessionSnapshotPublicationStatus, + ActiveSessionSnapshotRestore, ActiveSessionSnapshotResyncReason, + ActiveSessionSnapshotStorageLayoutKind, FeatureSupport, + validate_active_session_snapshot_restore, + }, + sharer, viewer, +}; + +fn assert_viewer_json_golden_round_trip(json: &str) -> viewer::DownstreamMessage { + let expected: serde_json::Value = serde_json::from_str(json).expect("valid fixture"); + let message = viewer::DownstreamMessage::from_json(json).expect("protocol decode"); + let actual: serde_json::Value = + serde_json::from_str(&message.to_json().expect("protocol encode")).expect("encoded JSON"); + assert_eq!(actual, expected); + message +} + +fn assert_sharer_json_golden_round_trip(json: &str) -> sharer::UpstreamMessage { + let expected: serde_json::Value = serde_json::from_str(json).expect("valid fixture"); + let message = sharer::UpstreamMessage::from_json(json).expect("protocol decode"); + let actual: serde_json::Value = + serde_json::from_str(&message.to_json().expect("protocol encode")).expect("encoded JSON"); + assert_eq!(actual, expected); + message +} + +fn capabilities() -> ActiveSessionSnapshotCapabilities { + ActiveSessionSnapshotCapabilities { + snapshot_schema_versions: vec![ACTIVE_RESTORE_SNAPSHOT_SCHEMA_VERSION_V1], + reducer_versions: vec![ACTIVE_RESTORE_REDUCER_VERSION_V1], + storage_layouts: vec![ActiveSessionSnapshotStorageLayoutKind::MonolithicV1], + compression_codecs: vec![ActiveSessionSnapshotCompression::Zstd], + } +} + +#[test] +fn legacy_rejoin_and_feature_support_json_are_unchanged() { + let message = assert_viewer_json_golden_round_trip(include_str!( + "fixtures/rejoined_full_terminal_legacy.json" + )); + assert!(!message.requires_active_session_snapshot_support()); + + match message { + viewer::DownstreamMessage::RejoinedSuccessfully { + active_session_snapshot_restore, + .. + } => assert!(active_session_snapshot_restore.is_none()), + _ => panic!("wrong message"), + } + + assert_eq!( + serde_json::to_value(FeatureSupport::default()).expect("feature support"), + serde_json::json!({ + "supports_agent_view": false, + "supports_full_role": false, + "supports_full_role_for_real": false + }) + ); +} + +#[test] +fn monolithic_v1_bootstrap_golden_has_an_exact_negotiated_contract() { + let message = + assert_viewer_json_golden_round_trip(include_str!("fixtures/rejoined_monolithic_v1.json")); + assert!(message.requires_active_session_snapshot_support()); + + let viewer::DownstreamMessage::RejoinedSuccessfully { + active_session_snapshot_restore: Some(restore), + .. + } = message + else { + panic!("wrong message"); + }; + + assert!(validate_active_session_snapshot_restore(&capabilities(), &restore).is_ok()); + let restore = *restore; + let ActiveSessionSnapshotRestore::Bootstrap { + snapshot, + resume_from_event_no, + catch_up_through_event_no, + retained_event_floor, + .. + } = restore + else { + panic!("expected bootstrap"); + }; + assert_eq!(snapshot.through_event_no, 41); + assert_eq!(resume_from_event_no, 42); + assert_eq!(catch_up_through_event_no, 45); + assert_eq!(retained_event_floor, 42); +} + +#[test] +fn prepared_receipt_is_outside_ordered_event_numbering() { + let message = + assert_sharer_json_golden_round_trip(include_str!("fixtures/publish_monolithic_v1.json")); + + let sharer::UpstreamMessage::PublishActiveSessionSnapshot { receipt } = message else { + panic!("wrong message"); + }; + assert_eq!(receipt.descriptor.through_event_no, 41); + assert_eq!(receipt.descriptor.snapshot_id, "snapshot-42"); + assert!(!format!("{receipt:?}").contains("opaque-prepared-receipt")); +} + +#[test] +fn all_publication_acknowledgements_round_trip() { + let statuses = [ + ActiveSessionSnapshotPublicationStatus::Prepared, + ActiveSessionSnapshotPublicationStatus::Committed, + ActiveSessionSnapshotPublicationStatus::Idempotent, + ActiveSessionSnapshotPublicationStatus::Stale, + ActiveSessionSnapshotPublicationStatus::NotContiguousYet, + ActiveSessionSnapshotPublicationStatus::Conflict, + ActiveSessionSnapshotPublicationStatus::TooLarge, + ActiveSessionSnapshotPublicationStatus::UnsupportedVersion, + ActiveSessionSnapshotPublicationStatus::InvalidObject, + ]; + + for status in statuses { + let message = sharer::DownstreamMessage::ActiveSessionSnapshotPublicationAck( + ActiveSessionSnapshotPublicationAck { + snapshot_id: "snapshot-42".into(), + status, + highest_contiguous_event_no: Some(40), + retained_event_floor: Some(1), + }, + ); + assert!(message.requires_active_session_snapshot_support()); + let encoded = message.to_json().expect("protocol encode"); + let decoded = sharer::DownstreamMessage::from_json(&encoded).expect("protocol decode"); + let sharer::DownstreamMessage::ActiveSessionSnapshotPublicationAck(ack) = decoded else { + panic!("wrong message"); + }; + assert_eq!(ack.status, status); + } +} + +#[test] +fn classified_resync_round_trips_and_unknown_reasons_fail_closed() { + let message = viewer::DownstreamMessage::ActiveSessionSnapshotResyncRequired { + reason: ActiveSessionSnapshotResyncReason::CursorBelowRetainedFloor { + retained_event_floor: 42, + }, + }; + assert!(message.requires_active_session_snapshot_support()); + let encoded = message.to_json().expect("protocol encode"); + assert!(matches!( + viewer::DownstreamMessage::from_json(&encoded).expect("protocol decode"), + viewer::DownstreamMessage::ActiveSessionSnapshotResyncRequired { + reason: ActiveSessionSnapshotResyncReason::CursorBelowRetainedFloor { + retained_event_floor: 42 + } + } + )); + assert!( + viewer::DownstreamMessage::from_json( + r#"{"ActiveSessionSnapshotResyncRequired":{"reason":"NewerUnknownReason"}}"# + ) + .is_err() + ); +} + +#[test] +fn unknown_layout_and_unadvertised_versions_fail_closed() { + assert!( + viewer::DownstreamMessage::from_json( + &include_str!("fixtures/rejoined_monolithic_v1.json") + .replace("\"MonolithicV1\"", "\"ChunkedV2\"") + ) + .is_err() + ); + + let message = + assert_viewer_json_golden_round_trip(include_str!("fixtures/rejoined_monolithic_v1.json")); + let viewer::DownstreamMessage::RejoinedSuccessfully { + active_session_snapshot_restore: Some(restore), + .. + } = message + else { + panic!("wrong message"); + }; + let unsupported = ActiveSessionSnapshotCapabilities { + snapshot_schema_versions: vec![2], + ..capabilities() + }; + assert!(validate_active_session_snapshot_restore(&unsupported, &restore).is_err()); +} diff --git a/tests/fixtures/publish_monolithic_v1.json b/tests/fixtures/publish_monolithic_v1.json new file mode 100644 index 0000000..92fc323 --- /dev/null +++ b/tests/fixtures/publish_monolithic_v1.json @@ -0,0 +1,32 @@ +{ + "PublishActiveSessionSnapshot": { + "receipt": { + "receipt": "opaque-prepared-receipt", + "descriptor": { + "snapshot_id": "snapshot-42", + "identity": { + "session_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "conversation_id": "conversation", + "execution_id": "314159", + "run_id": "run" + }, + "snapshot_schema_version": 1, + "reducer_version": 1, + "through_event_no": 41, + "captured_at_unix_ms": 1700000000000, + "storage": { + "MonolithicV1": { + "object": { + "object_id": "active-snapshots/snapshot-42", + "object_generation": "7", + "sha256": "0000000000000000000000000000000000000000000000000000000000000000", + "compressed_size_bytes": 512, + "uncompressed_size_bytes": 1024, + "compression_codec": "Zstd" + } + } + } + } + } + } +} diff --git a/tests/fixtures/rejoined_full_terminal_legacy.json b/tests/fixtures/rejoined_full_terminal_legacy.json new file mode 100644 index 0000000..53fd048 --- /dev/null +++ b/tests/fixtures/rejoined_full_terminal_legacy.json @@ -0,0 +1,24 @@ +{ + "RejoinedSuccessfully": { + "participant_list": { + "sharer": { + "info": { + "id": "sharer", + "profile_data": { + "firebase_uid": "uid", + "display_name": "Sharer", + "photo_url": null, + "email": null, + "input_replica_id": "replica" + }, + "selection": "None" + } + }, + "viewers": [], + "present_viewers": [], + "absent_viewers": [], + "guests": [], + "pending_guests": [] + } + } +} diff --git a/tests/fixtures/rejoined_monolithic_v1.json b/tests/fixtures/rejoined_monolithic_v1.json new file mode 100644 index 0000000..c4d7af1 --- /dev/null +++ b/tests/fixtures/rejoined_monolithic_v1.json @@ -0,0 +1,70 @@ +{ + "RejoinedSuccessfully": { + "participant_list": { + "sharer": { + "info": { + "id": "sharer", + "profile_data": { + "firebase_uid": "uid", + "display_name": "Sharer", + "photo_url": null, + "email": null, + "input_replica_id": "replica" + }, + "selection": "None" + } + }, + "viewers": [], + "present_viewers": [], + "absent_viewers": [], + "guests": [], + "pending_guests": [] + }, + "active_session_snapshot_restore": { + "Bootstrap": { + "negotiated": { + "protocol": { + "snapshot_schema_version": 1, + "reducer_version": 1, + "storage_layout": "MonolithicV1", + "compression_codec": "Zstd" + }, + "identity": { + "session_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "conversation_id": "conversation", + "execution_id": "314159", + "run_id": "run" + } + }, + "snapshot": { + "snapshot_id": "snapshot-42", + "identity": { + "session_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "conversation_id": "conversation", + "execution_id": "314159", + "run_id": "run" + }, + "snapshot_schema_version": 1, + "reducer_version": 1, + "through_event_no": 41, + "captured_at_unix_ms": 1700000000000, + "storage": { + "MonolithicV1": { + "object": { + "object_id": "active-snapshots/snapshot-42", + "object_generation": "7", + "sha256": "0000000000000000000000000000000000000000000000000000000000000000", + "compressed_size_bytes": 512, + "uncompressed_size_bytes": 1024, + "compression_codec": "Zstd" + } + } + } + }, + "resume_from_event_no": 42, + "catch_up_through_event_no": 45, + "retained_event_floor": 42 + } + } + } +}