| 1 | //! The message types that are communicated between the |
| 2 | //! server and a viewer client. |
| 3 | //! |
| 4 | //! When a client wants to join a shared session, the client |
| 5 | //! will make a request against /sessions/join/:uuid. The client |
| 6 | //! must then send an [`Initialize`] message with the relevant data |
| 7 | //! to join the shared session. If successful, the server |
| 8 | //! will acknowledge the joining of the shared |
| 9 | //! session via the [`JoinedSuccessfully`] message. |
| 10 | //! |
| 11 | //! To reconnect to a shared session, the viewer can use the same |
| 12 | //! join endpoint, setting the init payload appropriately. |
| 13 | //! The server will acknowledge the rejoining via the |
| 14 | //! [`RejoinedSuccessfully`] message. |
| 15 | |
| 16 | use crate::{ |
| 17 | common::{ |
| 18 | ActivePrompt, ActivePromptUpdate, AgentAttachment, AgentPromptFailureReason, |
| 19 | AgentPromptRequest, AgentPromptRequestId, BlockId, BufferId, CommandExecutionFailureReason, |
| 20 | CommandExecutionRequestId, ControlAction, ControlActionFailureReason, FeatureSupport, |
| 21 | InputOperationId, InputReplicaId, InputUpdate, InputUpdateFailureReason, |
| 22 | LinkAccessLevelUpdateResponse, OrderedTerminalEvent, ParticipantId, ParticipantList, |
| 23 | ParticipantPresenceUpdate, Role, RoleRequestId, RoleRequestResponse, Scrollback, |
| 24 | SelectionUpdate, TeamAccessLevelUpdateResponse, TeamAclData, TelemetryContext, |
| 25 | UniversalDeveloperInputContext, UniversalDeveloperInputContextUpdate, UserID, WindowSize, |
| 26 | WriteToPtyFailureReason, WriteToPtyRequestId, |
| 27 | }, |
| 28 | sharer::{self, LegacySessionSourceType, SessionSourceType}, |
| 29 | }; |
| 30 | use byte_unit::Byte; |
| 31 | use serde::{Deserialize, Serialize}; |
| 32 | |
| 33 | #[derive(Serialize, Deserialize, Clone, Copy, Debug)] |
| 34 | /// Sent by sharer client or server |
| 35 | /// when the shared session has been ended. |
| 36 | pub enum SessionEndedReason { |
| 37 | /// Unexpected, means something went wrong in the server. |
| 38 | InternalServerError, |
| 39 | /// The session was ended gracefully. |
| 40 | EndedBySharer, |
| 41 | /// The sharer was idle for too long. |
| 42 | InactivityLimitReached, |
| 43 | |
| 44 | /// DEPRECATED |
| 45 | ExceededSizeLimit, |
| 46 | } |
| 47 | |
| 48 | #[derive(Serialize, Deserialize, Clone, Debug)] |
| 49 | pub enum FailedToJoinReason { |
| 50 | /// Unexpected, means something went wrong in the server. |
| 51 | Invalid, |
| 52 | SessionNotFound, |
| 53 | WrongPassword, |
| 54 | InternalServerError, |
| 55 | MaxNumberOfParticipantsReached, |
| 56 | SessionNotAccessible, |
| 57 | } |
| 58 | |
| 59 | #[derive(Default, Debug, Deserialize, Serialize, Clone, Copy)] |
| 60 | pub enum RoleUpdatedReason { |
| 61 | #[default] |
| 62 | UpdatedBySharer, |
| 63 | InactivityLimitReached, |
| 64 | } |
| 65 | |
| 66 | impl From<sharer::RoleUpdateReason> for RoleUpdatedReason { |
| 67 | fn from(value: sharer::RoleUpdateReason) -> Self { |
| 68 | match value { |
| 69 | sharer::RoleUpdateReason::UpdatedBySharer => RoleUpdatedReason::UpdatedBySharer, |
| 70 | sharer::RoleUpdateReason::InactivityLimitReached => { |
| 71 | RoleUpdatedReason::InactivityLimitReached |
| 72 | } |
| 73 | } |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | #[derive(Serialize, Deserialize, Clone, Copy, Debug)] |
| 78 | pub enum ViewerRemovedReason { |
| 79 | LostAccess, |
| 80 | } |
| 81 | |
| 82 | /// The initial state that the viewer must supply when joining or rejoining |
| 83 | /// a shared session. |
| 84 | #[derive(Debug, Deserialize, Serialize)] |
| 85 | pub struct InitPayload { |
| 86 | /// The ID previously assigned to this viewer when joining. |
| 87 | /// Should be specified if the viewer is rejoining. |
| 88 | pub viewer_id: Option<ParticipantId>, |
| 89 | pub user_id: UserID, |
| 90 | /// If the viewer is reconnecting, they should specify the last event no received. |
| 91 | /// The server will only send events after this event no. |
| 92 | pub last_received_event_no: Option<usize>, |
| 93 | |
| 94 | /// The ID of the last block the viewer has seen. |
| 95 | /// Should only be specified when re-joining. |
| 96 | pub latest_block_id: Option<BlockId>, |
| 97 | |
| 98 | pub telemetry_context: Option<TelemetryContext>, |
| 99 | |
| 100 | /// Client feature support declaration. |
| 101 | #[serde(default)] |
| 102 | pub feature_support: FeatureSupport, |
| 103 | } |
| 104 | |
| 105 | /// The possible messages sent from server to client (viewer). |
| 106 | #[derive(Serialize, Deserialize, Clone)] |
| 107 | pub enum DownstreamMessage { |
| 108 | /// The server sends this message when the session was successfully joined. |
| 109 | /// TODO: add initial state to pass to viewer (e.g. size info) |
| 110 | JoinedSuccessfully { |
| 111 | scrollback: Box<Scrollback>, |
| 112 | /// The sharer's active prompt. |
| 113 | active_prompt: ActivePrompt, |
| 114 | /// The latest event no of the session the viewer will be catching up to. |
| 115 | /// If None, there are no events to catch up to. |
| 116 | latest_event_no: Option<usize>, |
| 117 | window_size: WindowSize, |
| 118 | |
| 119 | participant_list: Box<ParticipantList>, |
| 120 | |
| 121 | /// The ID assigned to this viewer |
| 122 | viewer_id: ParticipantId, |
| 123 | /// The Firebase UID assigned to this viewer. |
| 124 | viewer_firebase_uid: String, |
| 125 | |
| 126 | /// The block ID of the first block after scrollback. |
| 127 | /// The viewer can use this to identify buffer updates for |
| 128 | /// the first block. |
| 129 | init_block_id: BlockId, |
| 130 | |
| 131 | input_replica_id: InputReplicaId, |
| 132 | |
| 133 | /// The universal developer input context (model selection, etc.). |
| 134 | #[serde(default)] |
| 135 | universal_developer_input_context: Option<UniversalDeveloperInputContext>, |
| 136 | |
| 137 | /// The legacy source type for this shared session (i.e. user or ambient agent). |
| 138 | #[serde(default)] |
| 139 | #[deprecated(note = "please use `detailed_source_type` instead")] |
| 140 | source_type: LegacySessionSourceType, |
| 141 | |
| 142 | /// The detailed source type for this shared session. |
| 143 | #[serde(default)] |
| 144 | detailed_source_type: SessionSourceType, |
| 145 | |
| 146 | /// Optional orchestrator `task_id` carried alongside the source |
| 147 | /// type, mirroring `sharer::InitPayload::source_task_id`. Lets |
| 148 | /// viewers find this share's orchestrator task without keying |
| 149 | /// off the source-type variant kind. |
| 150 | #[serde(default)] |
| 151 | source_task_id: Option<String>, |
| 152 | }, |
| 153 | |
| 154 | /// The server sends this message when the session was successfully rejoined. |
| 155 | RejoinedSuccessfully { |
| 156 | participant_list: Box<ParticipantList>, |
| 157 | }, |
| 158 | |
| 159 | /// Sent when the viewer fails to join the shared session. |
| 160 | /// The client should not expect any more messages after this. |
| 161 | FailedToJoin { reason: FailedToJoinReason }, |
| 162 | |
| 163 | /// Sent when the shared session has been ended. |
| 164 | /// The client should not expect any more messages after this. |
| 165 | SessionEnded { reason: SessionEndedReason }, |
| 166 | |
| 167 | /// Update to the sharer's active prompt. |
| 168 | ActivePromptUpdated(ActivePromptUpdate), |
| 169 | |
| 170 | /// Update to the universal developer input context (model selection, etc.) from sharer or editor viewers. |
| 171 | UniversalDeveloperInputContextUpdated(UniversalDeveloperInputContextUpdate), |
| 172 | |
| 173 | /// Sent when there is any ordered terminal event. |
| 174 | /// These messages are only sent _after_ [`DownstreamMessage::JoinedSuccessfully`]. |
| 175 | OrderedTerminalEvent(OrderedTerminalEvent), |
| 176 | |
| 177 | /// Sent when the list of participants in the shared session changes. |
| 178 | ParticipantListUpdated(ParticipantList), |
| 179 | |
| 180 | /// Sent when a participant's presence changes. |
| 181 | ParticipantPresenceUpdated(ParticipantPresenceUpdate), |
| 182 | |
| 183 | /// The server has acknowledged the role request and sent it to the sharer. |
| 184 | /// There can only be at most one role request in flight per participant. |
| 185 | RoleRequestInFlight(RoleRequestId), |
| 186 | |
| 187 | /// The viewer's role request was responded to. |
| 188 | RoleRequestResponse(RoleRequestResponse), |
| 189 | |
| 190 | /// A participant's (identified by `participant_id`) role was updated. |
| 191 | ParticipantRoleChanged { |
| 192 | participant_id: ParticipantId, |
| 193 | reason: RoleUpdatedReason, |
| 194 | role: Role, |
| 195 | }, |
| 196 | |
| 197 | /// The input was updated by a participant. |
| 198 | /// When we receive our own update, we can treat it as an ack. |
| 199 | InputUpdated(InputUpdate), |
| 200 | |
| 201 | /// An input operation was rejected and should be undone. |
| 202 | InputUpdateRejected { |
| 203 | id: InputOperationId, |
| 204 | reason: InputUpdateFailureReason, |
| 205 | }, |
| 206 | |
| 207 | /// The server has acknowledged the command execution request and sent it to the sharer. |
| 208 | /// There can only be at most one command execution request in flight per participant. |
| 209 | CommandExecutionRequestInFlight(CommandExecutionRequestId), |
| 210 | |
| 211 | /// The viewer's command execution request failed. |
| 212 | /// Note: there is no "success" response; that is implicitly handled |
| 213 | /// by the fact that the command is executed. |
| 214 | CommandExecutionRequestFailed { |
| 215 | id: CommandExecutionRequestId, |
| 216 | reason: CommandExecutionFailureReason, |
| 217 | }, |
| 218 | |
| 219 | /// The viewer's write to pty request failed. |
| 220 | WriteToPtyRequestFailed { reason: WriteToPtyFailureReason }, |
| 221 | |
| 222 | /// The server has acknowledged the agent prompt request and sent it to the sharer. |
| 223 | AgentPromptRequestInFlight(AgentPromptRequestId), |
| 224 | |
| 225 | /// The viewer's agent prompt request failed. |
| 226 | /// Note: there is no "success" response; that is implicitly handled |
| 227 | /// by the fact that agent response events start streaming. |
| 228 | AgentPromptRequestFailed { reason: AgentPromptFailureReason }, |
| 229 | |
| 230 | /// The viewer's control action request failed. |
| 231 | ControlActionRequestFailed { reason: ControlActionFailureReason }, |
| 232 | |
| 233 | /// The viewer was removed from the session by the sharer. |
| 234 | ViewerRemoved { reason: ViewerRemovedReason }, |
| 235 | |
| 236 | /// Deprecated: superseded by [`DownstreamMessage::LinkAccessLevelUpdateResponse`]. |
| 237 | /// Kept temporarily for backward compatibility with older clients. Remove once |
| 238 | /// all clients handle `LinkAccessLevelUpdateResponse`. |
| 239 | LinkAccessLevelUpdated { role: Option<Role> }, |
| 240 | |
| 241 | /// Deprecated: superseded by [`DownstreamMessage::TeamAccessLevelUpdateResponse`]. |
| 242 | /// Kept temporarily for backward compatibility with older clients. Remove once |
| 243 | /// all clients handle `TeamAccessLevelUpdateResponse`. |
| 244 | TeamAccessLevelUpdated { |
| 245 | /// The UID of the updated team. |
| 246 | team_uid: String, |
| 247 | /// The ACL of the updated team. None if team has no ACL. |
| 248 | team_acl: Option<TeamAclData>, |
| 249 | }, |
| 250 | |
| 251 | /// The viewer's link access level update request was responded to. |
| 252 | LinkAccessLevelUpdateResponse(crate::common::LinkAccessLevelUpdateResponse), |
| 253 | |
| 254 | /// The request to add guests was responded to. |
| 255 | AddGuestsResponse(crate::common::AddGuestsResponse), |
| 256 | |
| 257 | /// The request to remove a guest was responded to. |
| 258 | RemoveGuestResponse(crate::common::RemoveGuestResponse), |
| 259 | |
| 260 | /// The request to update a pending user role was responded to. |
| 261 | UpdatePendingUserRoleResponse(crate::common::UpdatePendingUserRoleResponse), |
| 262 | |
| 263 | /// The viewer's team access level update request was responded to. |
| 264 | TeamAccessLevelUpdateResponse(crate::common::TeamAccessLevelUpdateResponse), |
| 265 | |
| 266 | /// A response to a [`UpstreamMessage::Ping`]. |
| 267 | /// Used to demonstrate that the server is still alive. |
| 268 | Pong { data: Vec<u8> }, |
| 269 | } |
| 270 | |
| 271 | impl DownstreamMessage { |
| 272 | pub fn from_json(json: &str) -> serde_json::Result<Self> { |
| 273 | serde_json::from_str(json) |
| 274 | } |
| 275 | |
| 276 | pub fn to_json(&self) -> serde_json::Result<String> { |
| 277 | serde_json::to_string(self) |
| 278 | } |
| 279 | |
| 280 | /// Downgrades all `Role::Full` fields to `Role::Executor`. |
| 281 | /// Used for backward compatibility with clients that don't support the Full role. |
| 282 | #[allow(deprecated)] |
| 283 | pub fn downgrade_full_roles(&mut self) { |
| 284 | match self { |
| 285 | Self::JoinedSuccessfully { |
| 286 | participant_list, .. |
| 287 | } => participant_list.downgrade_full_roles(), |
| 288 | Self::RejoinedSuccessfully { participant_list } => { |
| 289 | participant_list.downgrade_full_roles() |
| 290 | } |
| 291 | Self::ParticipantListUpdated(list) => list.downgrade_full_roles(), |
| 292 | Self::ParticipantRoleChanged { role, .. } => role.downgrade_full(), |
| 293 | Self::RoleRequestResponse(RoleRequestResponse::Approved { new_role }) => { |
| 294 | new_role.downgrade_full() |
| 295 | } |
| 296 | Self::LinkAccessLevelUpdated { role: Some(role) } => { |
| 297 | role.downgrade_full(); |
| 298 | } |
| 299 | Self::TeamAccessLevelUpdated { |
| 300 | team_acl: Some(team_acl), |
| 301 | .. |
| 302 | } => { |
| 303 | team_acl.acl.downgrade_full(); |
| 304 | } |
| 305 | Self::LinkAccessLevelUpdateResponse(LinkAccessLevelUpdateResponse::Ok { |
| 306 | role: Some(role), |
| 307 | }) => { |
| 308 | role.downgrade_full(); |
| 309 | } |
| 310 | Self::TeamAccessLevelUpdateResponse(TeamAccessLevelUpdateResponse::Success { |
| 311 | team_acl: Some(team_acl), |
| 312 | .. |
| 313 | }) => { |
| 314 | team_acl.acl.downgrade_full(); |
| 315 | } |
| 316 | _ => {} |
| 317 | } |
| 318 | } |
| 319 | } |
| 320 | |
| 321 | /// The possible messages sent from client (viewer) to server. |
| 322 | #[derive(Debug, Serialize, Deserialize)] |
| 323 | pub enum UpstreamMessage { |
| 324 | /// The client sends this message to join the shared session. |
| 325 | Initialize(InitPayload), |
| 326 | |
| 327 | /// A heartbeat message to demonstrate that the |
| 328 | /// client is still alive. |
| 329 | Ping { data: Vec<u8> }, |
| 330 | |
| 331 | /// Sent when the viewer changes what they have selected. |
| 332 | UpdateSelection(SelectionUpdate), |
| 333 | |
| 334 | /// The viewer is requesting a new role. |
| 335 | RequestRole(Role), |
| 336 | |
| 337 | /// The viewer no longer wants to change roles. |
| 338 | CancelRoleRequest(RoleRequestId), |
| 339 | |
| 340 | /// The viewer updated their input. |
| 341 | /// This is an optimistic update and thus was already applied on the viewer's client. |
| 342 | UpdateInput(InputUpdate), |
| 343 | |
| 344 | /// The viewer is requesting the sharer to execute the provided command |
| 345 | /// in the given buffer. |
| 346 | ExecuteCommand { |
| 347 | buffer_id: BufferId, |
| 348 | command: String, |
| 349 | }, |
| 350 | |
| 351 | /// The viewer is requesting to write to the pty, |
| 352 | /// specifically to a long running command. |
| 353 | WriteToPty { |
| 354 | request_id: WriteToPtyRequestId, |
| 355 | bytes: Vec<u8>, |
| 356 | }, |
| 357 | |
| 358 | /// The viewer is requesting to send an agent prompt. |
| 359 | /// If there's an existing in-flight request for the same conversation, |
| 360 | /// it will be cancelled and replaced with this new request. |
| 361 | SendAgentPrompt(AgentPromptRequest), |
| 362 | |
| 363 | /// The viewer (with Editor role) is updating the universal developer input context. |
| 364 | UpdateUniversalDeveloperInputContext(UniversalDeveloperInputContextUpdate), |
| 365 | |
| 366 | /// The viewer is requesting a one-off control action to be applied to the shared session. |
| 367 | SendControlAction(ControlAction), |
| 368 | |
| 369 | /// The viewer has reauthenticated. |
| 370 | Reauthenticated { user_id: UserID }, |
| 371 | |
| 372 | /// The viewer updated the session's link permissions. |
| 373 | UpdateLinkAccessLevel { role: Option<Role> }, |
| 374 | |
| 375 | /// The viewer updated the session's team permissions. |
| 376 | UpdateTeamAccessLevel { |
| 377 | team_uid: String, |
| 378 | role: Option<Role>, |
| 379 | }, |
| 380 | |
| 381 | /// The viewer added users as session guests by email. |
| 382 | AddGuests { emails: Vec<String>, role: Role }, |
| 383 | |
| 384 | /// The viewer removed a user as a session guest. |
| 385 | RemoveGuest { user_uid: String }, |
| 386 | |
| 387 | /// The viewer removed a pending user as a session guest. |
| 388 | RemovePendingGuest { email: String }, |
| 389 | |
| 390 | /// The viewer changed a user's role. |
| 391 | UpdateUserRole { user_uid: String, role: Role }, |
| 392 | |
| 393 | /// The viewer changed a pending user's role. |
| 394 | UpdatePendingUserRole { email: String, role: Role }, |
| 395 | |
| 396 | /// The viewer is reporting its terminal size to the sharer. |
| 397 | /// Used for remote-control sessions where the viewer's viewport should drive the PTY size. |
| 398 | ReportTerminalSize { window_size: WindowSize }, |
| 399 | } |
| 400 | |
| 401 | impl UpstreamMessage { |
| 402 | pub fn from_json(json: &str) -> serde_json::Result<Self> { |
| 403 | serde_json::from_str(json) |
| 404 | } |
| 405 | |
| 406 | pub fn to_json(&self) -> serde_json::Result<String> { |
| 407 | serde_json::to_string(self) |
| 408 | } |
| 409 | |
| 410 | pub fn num_bytes(&self) -> Byte { |
| 411 | match self { |
| 412 | UpstreamMessage::UpdateInput(input_update) => input_update.num_bytes(), |
| 413 | UpstreamMessage::ExecuteCommand { command, .. } => command.len().into(), |
| 414 | UpstreamMessage::WriteToPty { bytes, .. } => bytes.len().into(), |
| 415 | UpstreamMessage::SendAgentPrompt(request) => { |
| 416 | // Count prompt length + attachments |
| 417 | let prompt_bytes: Byte = request.prompt.len().into(); |
| 418 | let attachments_bytes: Byte = request |
| 419 | .attachments |
| 420 | .iter() |
| 421 | .map(|att| match att { |
| 422 | AgentAttachment::PlainText { content } => content.len(), |
| 423 | // Block's are already included in the shared session thus far, |
| 424 | // so we do not have to count them again here. |
| 425 | AgentAttachment::BlockReference { .. } => 0, |
| 426 | // FileReference is just IDs — the actual data is in GCS. |
| 427 | AgentAttachment::FileReference { .. } => 0, |
| 428 | }) |
| 429 | .sum::<usize>() |
| 430 | .into(); |
| 431 | prompt_bytes |
| 432 | .add(attachments_bytes) |
| 433 | .unwrap_or(u64::MAX.into()) |
| 434 | } |
| 435 | _ => Byte::from_u64(0), |
| 436 | } |
| 437 | } |
| 438 | } |