main
rs 629 lines 21.4 KB
Raw
1 //! The message types that are communicated between the
2 //! server and a sharer client.
3 //!
4 //! When a client wants to create a shared session, the client
5 //! will make a request against /sessions/create. The client
6 //! must then send an [`Initialize`] message with the relevant data
7 //! to start the shared session. If successful, the server
8 //! will acknowledge the creation of the shared
9 //! session via the [`SessionInitialized`] message.
10 //!
11 //! Remember to annotate #[serde(default)] to every new field added for backward compatibility,
12 //! since old clients may not specify new fields expected by the server.
13
14 use crate::common::{
15 ActivePrompt, ActivePromptUpdate, AgentPromptFailureReason, AgentPromptRequest,
16 AgentPromptRequestId, BlockId, BufferId, CommandExecutionFailureReason,
17 CommandExecutionRequestId, ControlAction, ControlActionFailureReason, ControlActionRequestId,
18 FeatureSupport, InputOperationId, InputReplicaId, InputUpdate, InputUpdateFailureReason,
19 OrderedTerminalEvent, ParticipantId, ParticipantList, ParticipantPresenceUpdate, Role,
20 RoleRequestId, RoleRequestResponse, Selection, SelectionUpdate, SessionId, SessionSecret,
21 TelemetryContext, UniversalDeveloperInputContext, UniversalDeveloperInputContextUpdate, UserID,
22 WindowSize, WriteToPtyFailureReason, WriteToPtyRequestId,
23 };
24
25 use super::common::Scrollback;
26 use byte_unit::Byte;
27 use serde::{Deserialize, Deserializer, Serialize};
28 use uuid::Uuid;
29
30 /// Possible reasons why the server might gracefully terminate
31 /// a shared session.
32 #[derive(Clone, Serialize, Deserialize, Debug)]
33 pub enum SessionTerminatedReason {
34 /// Unknown error occurred. Session cannot continue.
35 InternalServerError {
36 /// Details about what happened. This should
37 /// 1. only be provided to the sharer client,
38 /// 2. not necessarily be user-facing, and
39 /// 3. clients should _not_ try to match on the exact message
40 details: String,
41 },
42 /// The session exceeded its size limit.
43 ExceededSizeLimit,
44 /// The user does not have any more quota remaining.
45 NoUserQuotaRemaining {
46 // This is left as an empty struct to make it
47 // easier to add fields (e.g. next refresh time)
48 // in the future in a backwards-compatible way.
49 },
50 }
51
52 impl SessionTerminatedReason {
53 pub fn internal_server_error(details: impl Into<String>) -> Self {
54 Self::InternalServerError {
55 details: details.into(),
56 }
57 }
58 pub fn internal_server_error_without_details() -> Self {
59 Self::internal_server_error(String::new())
60 }
61 }
62
63 #[derive(Serialize, Deserialize, Clone, Copy, Debug)]
64 /// Client-side reasons for ending the shared session.
65 pub enum SessionEndedReason {
66 /// The session was ended gracefully.
67 EndedBySharer,
68 /// The sharer was idle for too long.
69 InactivityLimitReached,
70 /// The session exceeded its size limit.
71 // TODO: remove as part of quota enforcement work
72 ExceededSizeLimit,
73 }
74
75 #[derive(Deserialize, Serialize, Debug)]
76 pub enum ReconnectionFailedReason {
77 /// Unexpected, means something went wrong in the server.
78 Invalid,
79 /// The session with the specified ID does not exist.
80 SessionNotFound,
81 /// The specified password was incorrect.
82 WrongPassword,
83 /// The specified reconnection token was incorrect.
84 WrongReconnectionToken,
85 /// The firebase ID of the sharer was missing or doesn't match the original one.
86 WrongFirebaseUid,
87 /// The sharer does not have any remaining quota.
88 NoUserQuotaRemaining,
89 /// The session is not accessible.
90 SessionNotAccessible,
91 }
92
93 #[derive(Default, Debug, Deserialize, Serialize, Clone, Copy)]
94 pub enum RoleUpdateReason {
95 #[default]
96 UpdatedBySharer,
97 InactivityLimitReached,
98 }
99
100 #[derive(Debug, Deserialize, Serialize, Clone, Copy)]
101 pub enum QuotaType {
102 BytesUsed,
103 SessionsCreated,
104 }
105
106 /// The reasons we might fail to initialize a new session.
107 #[derive(Clone, Serialize, Deserialize, Debug)]
108 pub enum FailedToInitializeSessionReason {
109 /// The scrollback exceeds the user's quota.
110 ScrollbackTooLarge {
111 // This is left as an empty struct to make it
112 // easier to add fields (e.g. remaining scrollback size)
113 // in the future in a backwards-compatible way.
114 },
115 /// The sharer does not have any remaining quota.
116 NoUserQuotaRemaining { quota_type: QuotaType },
117 /// The sharer could not be attributed to a Warp user.
118 UserNotFound,
119 /// Something unexpectedly went wrong.
120 InternalServerError {
121 /// Details about what happened. This should
122 /// not necessarily be user-facing, and clients should
123 /// _not_ try to match on the exact message.
124 details: String,
125 },
126 }
127
128 impl FailedToInitializeSessionReason {
129 pub fn internal_server_error_without_details() -> Self {
130 FailedToInitializeSessionReason::InternalServerError {
131 details: String::new(),
132 }
133 }
134 }
135
136 // Permission response types are now in common::permissions.
137 // Re-exported here for backward compatibility.
138 pub use crate::common::{
139 AddGuestsResponse, FailedToAddGuestsReason, FailedToRemoveGuestReason,
140 FailedToUpdatePendingUserRoleReason, FailedToUpdateTeamAccessLevelReason,
141 LinkAccessLevelUpdateResponse, RemoveGuestResponse, TeamAccessLevelUpdateResponse,
142 UpdatePendingUserRoleResponse,
143 };
144
145 #[derive(Clone, Debug, Serialize, Default)]
146 pub enum SessionSourceType {
147 /// The session was started by a user directly.
148 #[default]
149 User,
150 /// The session was started in the course of spinning up an ambient agent.
151 AmbientAgent {
152 #[serde(default)]
153 task_id: Option<String>,
154 },
155 }
156
157 /// Mirrors the legacy unit-variant form and the new struct-variant form so
158 /// the custom `Deserialize` impl can accept both shapes.
159 #[derive(Deserialize)]
160 #[serde(untagged)]
161 enum SessionSourceTypeWire {
162 /// Legacy representation: bare `"User"` or `"AmbientAgent"`.
163 Legacy(LegacySessionSourceType),
164 /// New representation: externally tagged `AmbientAgent` with fields, e.g.
165 /// `{ "AmbientAgent": { "task_id": "..." } }`.
166 New {
167 #[serde(rename = "AmbientAgent")]
168 ambient_agent: AmbientAgentFields,
169 },
170 }
171
172 #[derive(Deserialize)]
173 struct AmbientAgentFields {
174 #[serde(default)]
175 task_id: Option<String>,
176 }
177
178 impl From<SessionSourceTypeWire> for SessionSourceType {
179 fn from(value: SessionSourceTypeWire) -> Self {
180 match value {
181 SessionSourceTypeWire::Legacy(LegacySessionSourceType::User) => SessionSourceType::User,
182 SessionSourceTypeWire::Legacy(LegacySessionSourceType::AmbientAgent) => {
183 SessionSourceType::AmbientAgent { task_id: None }
184 }
185 SessionSourceTypeWire::New {
186 ambient_agent: AmbientAgentFields { task_id },
187 } => SessionSourceType::AmbientAgent { task_id },
188 }
189 }
190 }
191
192 impl<'de> Deserialize<'de> for SessionSourceType {
193 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
194 where
195 D: Deserializer<'de>,
196 {
197 let wire = SessionSourceTypeWire::deserialize(deserializer)?;
198 Ok(SessionSourceType::from(wire))
199 }
200 }
201
202 #[derive(Clone, Debug, Serialize, Deserialize, Default)]
203 pub enum LegacySessionSourceType {
204 #[default]
205 User,
206 AmbientAgent,
207 }
208
209 impl From<&SessionSourceType> for LegacySessionSourceType {
210 fn from(value: &SessionSourceType) -> Self {
211 match value {
212 SessionSourceType::User => LegacySessionSourceType::User,
213 SessionSourceType::AmbientAgent { .. } => LegacySessionSourceType::AmbientAgent,
214 }
215 }
216 }
217
218 /// Configures the lifetime of the session after sharing ends.
219 #[derive(Serialize, Deserialize, Clone, Copy, Debug, Default)]
220 pub enum Lifetime {
221 /// The session is deleted immediately when sharing ends.
222 #[default]
223 Ephemeral,
224 /// The session persists after sharing ends.
225 ///
226 /// It is not specified how long a lingering session is available for after it ends.
227 /// Currently, all session contents expire after one week, but this is a server implementation
228 /// detail that clients must not rely on. In the future, we may expose a lifetime option that
229 /// includes a client-provided TTL.
230 Lingering,
231 }
232
233 /// The reasons the sharer may request extended retention for a shared session.
234 #[derive(Serialize, Deserialize, Clone, Copy, Debug, Default)]
235 pub enum SessionRetentionReason {
236 /// Environment setup failed. The session should remain available so users can inspect setup
237 /// command output from the shared session link after the sandbox shuts down.
238 #[default]
239 SetupFailed,
240 }
241
242 /// The initial state that the sharer must supply when starting
243 /// a shared session.
244 #[derive(Debug, Deserialize, Serialize)]
245 pub struct InitPayload {
246 pub scrollback: Scrollback,
247
248 pub active_prompt: ActivePrompt,
249
250 pub window_size: WindowSize,
251
252 pub user_id: UserID,
253
254 /// What the sharer currently has selected for presence.
255 pub selection: Selection,
256
257 pub init_block_id: BlockId,
258
259 pub input_replica_id: InputReplicaId,
260
261 pub telemetry_context: Option<TelemetryContext>,
262
263 #[serde(default)]
264 pub lifetime: Lifetime,
265
266 /// The universal developer input context state.
267 #[serde(default)]
268 pub universal_developer_input_context: Option<UniversalDeveloperInputContext>,
269
270 /// The source type for this shared session (i.e. user or ambient agent).
271 #[serde(default)]
272 pub source_type: SessionSourceType,
273
274 /// Optional orchestrator `task_id` carried alongside `source_type`.
275 /// Set when the sharer wants downstream orchestration discovery to find
276 /// this share's children regardless of variant kind. Sidecar so the
277 /// `User` variant can stay a unit and old viewers ignore it.
278 #[serde(default)]
279 pub source_task_id: Option<String>,
280
281 /// Client feature support declaration.
282 #[serde(default)]
283 pub feature_support: FeatureSupport,
284 }
285
286 /// The reconnection token for a shared session.
287 /// A sharer must specify this to reconnect to the session and resume sharing.
288 /// The client should treat this as some opaque string.
289 #[derive(Hash, Serialize, Deserialize, Eq, PartialEq, Clone)]
290 #[serde(transparent)]
291 pub struct ReconnectToken(String);
292 impl ReconnectToken {
293 pub fn new() -> Self {
294 Self::default()
295 }
296 }
297
298 impl Default for ReconnectToken {
299 fn default() -> Self {
300 Self(Uuid::new_v4().to_string())
301 }
302 }
303
304 /// Override the Display impl for the token to return a mask.
305 /// This makes it harder to leak the token by accident.
306 impl std::fmt::Display for ReconnectToken {
307 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
308 write!(f, "***")
309 }
310 }
311
312 /// Override the Debug impl for the token to return a mask.
313 /// This makes it harder to leak the token by accident.
314 impl std::fmt::Debug for ReconnectToken {
315 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
316 write!(f, "***")
317 }
318 }
319
320 /// The `warp` server framework uses [`FromStr`] to deserialize
321 /// the string from the route.
322 impl std::str::FromStr for ReconnectToken {
323 type Err = core::convert::Infallible;
324 fn from_str(s: &str) -> Result<Self, Self::Err> {
325 String::from_str(s).map(ReconnectToken)
326 }
327 }
328
329 /// Payload sharer must supply to reconnect to an existing shared session
330 /// when the websocket was terminated by the server.
331 #[derive(Debug, Deserialize, Serialize)]
332 pub struct ReconnectPayload {
333 // TODO: Remove in favour of ACLs
334 pub session_secret: SessionSecret,
335 pub reconnect_token: ReconnectToken,
336
337 pub user_id: UserID,
338
339 /// The ID of the latest block when reconnecting.
340 /// This allows the sharer to catch up on any
341 /// missed input updates while they were disconnected.
342 pub latest_block_id: BlockId,
343
344 /// What the sharer currently has selected for presence.
345 pub selection: Selection,
346
347 /// Client feature support declaration.
348 #[serde(default)]
349 pub feature_support: FeatureSupport,
350 }
351
352 /// The possible messages sent from server to client (sharer).
353 #[derive(Deserialize, Serialize)]
354 pub enum DownstreamMessage {
355 /// The server sends this message when the session was successfully created.
356 SessionInitialized {
357 session_id: SessionId,
358 // TODO: Remove in favour of ACLs
359 session_secret: SessionSecret,
360 reconnect_token: ReconnectToken,
361 /// The ID assigned to the sharer
362 sharer_id: ParticipantId,
363 /// The Firebase UID assigned to the sharer.
364 sharer_firebase_uid: String,
365 },
366
367 /// The server denied the initialization request. No further messages will be processed.
368 FailedToInitializeSession {
369 reason: FailedToInitializeSessionReason,
370 },
371
372 /// The session was terminated. No further messages
373 /// will be processed.
374 SessionTerminated { reason: SessionTerminatedReason },
375
376 /// The server accepted the reconnection request.
377 SessionReconnected {
378 /// The last event no received by the server.
379 /// The sharer can use this to update the server with any newer events created while disconnected.
380 last_received_event_no: Option<usize>,
381 participant_list: ParticipantList,
382 },
383
384 /// The server denied the reconnection request. No further messages will be processed.
385 FailedToReconnect { reason: ReconnectionFailedReason },
386
387 /// The server sends this to confirm it has fully processed events up to the latest_processed_event_no,
388 /// and the sharer can safely remove them from memory.
389 EventsProcessedAck { latest_processed_event_no: usize },
390
391 /// Sent when the list of participants in the shared session changes.
392 ParticipantListUpdated(ParticipantList),
393
394 /// Sent when a participant's presence changes.
395 ParticipantPresenceUpdated(ParticipantPresenceUpdate),
396
397 /// The participant (identified by `participant_id`) requested the `role` role.
398 RoleRequested {
399 participant_id: ParticipantId,
400 request_id: RoleRequestId,
401 role: Role,
402 },
403
404 /// The participant (identified by `participant_id`) has cancelled their role request.
405 RoleRequestCancelled {
406 participant_id: ParticipantId,
407 request_id: RoleRequestId,
408 },
409
410 /// A participant's (identified by `participant_id`) role was updated.
411 ParticipantRoleChanged {
412 participant_id: ParticipantId,
413 role: Role,
414 },
415
416 /// A participant requested a control action (e.g. cancel conversation).
417 ControlActionRequested {
418 participant_id: ParticipantId,
419 request_id: ControlActionRequestId,
420 action: ControlAction,
421 },
422
423 /// The input was updated by a participant.
424 /// When we receive our own update, we can treat it as an ack.
425 InputUpdated(InputUpdate),
426
427 /// The rejection was successfully applied (does not need to be retried by the client).
428 InputUpdateRejectedAck { id: InputOperationId },
429
430 /// A participant requested that the given `command` be run in the given buffer.
431 CommandExecutionRequested {
432 id: CommandExecutionRequestId,
433 participant_id: ParticipantId,
434 buffer_id: BufferId,
435 command: String,
436 },
437
438 /// A participant requested to write to the pty, specifically for a long running command.
439 WriteToPtyRequested {
440 id: WriteToPtyRequestId,
441 bytes: Vec<u8>,
442 },
443
444 /// A participant requested to send an agent prompt.
445 AgentPromptRequested {
446 id: AgentPromptRequestId,
447 participant_id: ParticipantId,
448 request: AgentPromptRequest,
449 },
450
451 /// The sharer's link access level update request was responded to.
452 LinkAccessLevelUpdateResponse(LinkAccessLevelUpdateResponse),
453
454 /// The request to add guests was responded to.
455 AddGuestsResponse(AddGuestsResponse),
456
457 /// The request to remove a guest was responded to.
458 RemoveGuestResponse(RemoveGuestResponse),
459
460 /// The request to update a pending user role was responded to.
461 UpdatePendingUserRoleResponse(UpdatePendingUserRoleResponse),
462
463 /// The sharer's team access level update request was responsed to.
464 TeamAccessLevelUpdateResponse(TeamAccessLevelUpdateResponse),
465
466 /// Update to the universal developer input context from sharer or editor viewers.
467 UniversalDeveloperInputContextUpdated(UniversalDeveloperInputContextUpdate),
468
469 /// A viewer reported its terminal size.
470 /// Used for remote-control sessions where the viewer's viewport should drive the PTY size.
471 ViewerTerminalSizeReported {
472 participant_id: ParticipantId,
473 window_size: WindowSize,
474 },
475
476 /// A response to a [`UpstreamMessage::Ping`].
477 /// Used to demonstrate that the server is still alive.
478 Pong { data: Vec<u8> },
479 }
480
481 impl DownstreamMessage {
482 pub fn from_json(json: &str) -> serde_json::Result<Self> {
483 serde_json::from_str(json)
484 }
485
486 pub fn to_json(&self) -> serde_json::Result<String> {
487 serde_json::to_string(self)
488 }
489 }
490
491 /// The possible messages sent from client (sharer) to server.
492 // `Initialize(InitPayload)` is much larger than the other variants because
493 // `InitPayload` carries scrollback and feature-support data. Boxing it would
494 // be wire-compatible but churn every call site; suppress the lint instead.
495 #[allow(clippy::large_enum_variant)]
496 #[derive(Debug, Deserialize, Serialize)]
497 pub enum UpstreamMessage {
498 /// The client sends this message to start a shared session.
499 /// supplying any necessary initial state.
500 /// TODO: add size info, etc.
501 Initialize(InitPayload),
502
503 /// A heartbeat message to demonstrate that the
504 /// client is still alive.
505 Ping { data: Vec<u8> },
506
507 /// The client sends this message to explicitly end a session
508 /// and notify viewers before the websocket closes.
509 EndSession { reason: SessionEndedReason },
510
511 /// The client sends this message to request that the server retain session data longer than
512 /// normal after sharing ends.
513 ExtendSessionRetention {
514 #[serde(default)]
515 reason: SessionRetentionReason,
516 },
517
518 /// Update to the sharer's active prompt.
519 UpdateActivePrompt(ActivePromptUpdate),
520
521 /// Update to the universal developer input context (model selection, etc.).
522 UpdateUniversalDeveloperInputContext(UniversalDeveloperInputContextUpdate),
523
524 /// Sent when there is any ordered terminal event.
525 OrderedTerminalEvent(OrderedTerminalEvent),
526
527 /// Sent to reconnect to the server after disconnection.
528 Reconnect(ReconnectPayload),
529
530 /// Sent when the sharer changes what they have selected.
531 UpdateSelection(SelectionUpdate),
532
533 /// Changes the participant's (identified by `participant_id`) role.
534 UpdateRole {
535 participant_id: ParticipantId,
536 role: Role,
537 },
538
539 /// Changes the user's role (applied to all participants with the same UID).
540 UpdateUserRole { user_uid: String, role: Role },
541
542 /// Changes the pending user's role (applied to all participants with the same UID).
543 UpdatePendingUserRole { email: String, role: Role },
544
545 /// Responds to the participant's (identified by `participant_id`) role request.
546 RespondToRoleRequest {
547 participant_id: ParticipantId,
548 request_id: RoleRequestId,
549 response: RoleRequestResponse,
550 },
551
552 /// Updates all participants' roles to be [Role::Reader].
553 UpdateAllRolesToReader { reason: RoleUpdateReason },
554
555 /// The sharer updated the input.
556 UpdateInput(InputUpdate),
557
558 /// The given operation should be undone on all participants.
559 RejectInputUpdate {
560 id: InputOperationId,
561 reason: InputUpdateFailureReason,
562 },
563
564 /// The given command execution request was denied for the specified `reason`.
565 RejectCommandExecutionRequest {
566 id: CommandExecutionRequestId,
567 participant_id: ParticipantId,
568 reason: CommandExecutionFailureReason,
569 },
570
571 /// The given write to pty request was denied for the specified `reason`.
572 RejectWriteToPtyRequest {
573 id: WriteToPtyRequestId,
574 reason: WriteToPtyFailureReason,
575 },
576
577 /// The given agent prompt request was denied for the specified `reason`.
578 RejectAgentPromptRequest {
579 id: AgentPromptRequestId,
580 participant_id: ParticipantId,
581 reason: AgentPromptFailureReason,
582 },
583
584 /// The given control action request was denied for the specified `reason`.
585 RejectControlActionRequest {
586 participant_id: ParticipantId,
587 request_id: ControlActionRequestId,
588 reason: ControlActionFailureReason,
589 },
590
591 /// The sharer updated the session's link permissions.
592 UpdateLinkAccessLevel { role: Option<Role> },
593
594 /// The sharer updated the session's team permissions.
595 UpdateTeamAccessLevel {
596 team_uid: String,
597 role: Option<Role>,
598 },
599
600 /// The sharer added users as session guests by email.
601 AddGuests { emails: Vec<String>, role: Role },
602
603 /// The sharer removed a user as a session guest.
604 RemoveGuest { user_uid: String },
605
606 /// The sharer removed a pending user as a session guest.
607 RemovePendingGuest { email: String },
608 }
609
610 impl UpstreamMessage {
611 pub fn from_json(json: &str) -> serde_json::Result<Self> {
612 serde_json::from_str(json)
613 }
614
615 pub fn to_json(&self) -> serde_json::Result<String> {
616 serde_json::to_string(self)
617 }
618
619 pub fn num_bytes(&self) -> Byte {
620 match self {
621 UpstreamMessage::Initialize(init_payload) => init_payload.scrollback.num_bytes(),
622 UpstreamMessage::OrderedTerminalEvent(ordered_terminal_event) => {
623 ordered_terminal_event.num_bytes()
624 }
625 UpstreamMessage::UpdateInput(input_update) => input_update.num_bytes(),
626 _ => Byte::from_u64(0),
627 }
628 }
629 }