1use std::path::Path;
5use std::sync::Arc;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::time::{Duration, Instant};
8
9use axum::extract::ws::{Message, WebSocketUpgrade};
10use axum::extract::{ConnectInfo, State};
11use axum::http::{HeaderMap, StatusCode, header};
12use axum::response::{IntoResponse as _, Response};
13use futures_util::{SinkExt as _, StreamExt};
14use tokio::sync::{mpsc, oneshot};
15
16use base64::Engine as _;
17use forge_primitives::SessionSlot;
18use forge_primitives::browser::{
19 BrowserFrame, BrowserInputEvent, BrowserTabList, BrowserViewState, MouseAction,
20};
21use forge_primitives::document::DocumentRead;
22use forge_primitives::wire::COMPRESSED_JSON_TAG;
23use forge_workspace::browser::BrowserAttachment;
24use serde_json::json;
25
26use super::batch::{self, Batch};
27use super::envelope::{
28 ClientMessage, ClientSettings, SeatFrameWire, ServerMessage, SettingsRefusal, Subject,
29};
30use super::frame::FrameTransport as _;
31use super::origin as origin_gate;
32use super::wire::{
33 FileContentWire, OpenTimings, ReviewedThreadWire, cap_message_record, cap_tool_records,
34 conversation_for, encode_subject, encode_subject_timed, floor_cursor, page_within,
35 placement_wire,
36};
37use super::{PROTOCOL_VERSION, TransportState, WRITE_BOUND, WriteStalled, write_within};
38use crate::live::Live;
39use crate::surface::{DictateOutcome, ViewSurface};
40use crate::{Command, DispatchError, SeatUpdate, SessionUpdate};
41
42struct Holds<'a> {
49 surface: &'a Arc<ViewSurface>,
50 seats: Vec<SessionSlot>,
51}
52
53#[derive(Default)]
60struct Watching {
61 subjects: Vec<Subject>,
62 showing: Option<SessionSlot>,
63 showing_attached: bool,
72}
73
74impl<'a> Holds<'a> {
75 fn new(surface: &'a Arc<ViewSurface>) -> Self {
76 Self { surface, seats: Vec::new() }
77 }
78
79 fn take(&mut self, slot: &SessionSlot) {
81 self.seats.push(slot.clone());
82 }
83
84 fn give_back(&mut self, slot: &SessionSlot) -> bool {
86 let Some(at) = self.seats.iter().position(|held| held == slot) else {
87 return false;
88 };
89 self.seats.remove(at);
90 self.surface.release_seat(slot);
91 true
92 }
93}
94
95impl Drop for Holds<'_> {
96 fn drop(&mut self) {
97 for slot in &self.seats {
98 self.surface.release_seat(slot);
99 }
100 }
101}
102
103pub(crate) const HEARTBEAT_EVERY: Duration = Duration::from_secs(15);
111
112pub(crate) const HEARTBEAT_DEAD: Duration = Duration::from_secs(60);
118
119pub(crate) const WEDGE_DEAD: Duration = Duration::from_secs(240);
139
140#[derive(Debug)]
153struct Progress {
154 base: tokio::time::Instant,
157 at: std::sync::atomic::AtomicU64,
158 showing: std::sync::Mutex<Option<SessionSlot>>,
162}
163
164impl Progress {
165 fn new() -> Self {
166 Self {
167 base: tokio::time::Instant::now(),
168 at: std::sync::atomic::AtomicU64::new(0),
169 showing: std::sync::Mutex::new(None),
170 }
171 }
172
173 fn stamp(&self) {
175 let millis = u64::try_from(self.base.elapsed().as_millis()).unwrap_or(u64::MAX);
176 self.at.store(millis, Ordering::Relaxed);
177 }
178
179 fn silent_for(&self, now: tokio::time::Instant) -> Duration {
181 let last = Duration::from_millis(self.at.load(Ordering::Relaxed));
182 now.saturating_duration_since(self.base + last)
183 }
184
185 fn note_showing(&self, seat: Option<&SessionSlot>) {
187 *self.showing.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = seat.cloned();
188 }
189
190 fn showing_seat(&self) -> Option<SessionSlot> {
192 self.showing.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone()
193 }
194}
195
196#[derive(Debug)]
202struct WedgeStalled {
203 bound: Duration,
204 silent: Duration,
205}
206
207impl std::fmt::Display for WedgeStalled {
208 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
209 write!(
210 f,
211 "no write completed on the connection for {:?}, past its {:?} bound",
212 self.silent, self.bound
213 )
214 }
215}
216
217impl std::error::Error for WedgeStalled {}
218
219#[derive(Debug)]
226struct PeerVanished;
227
228impl std::fmt::Display for PeerVanished {
229 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
230 write!(f, "the peer never answered the keepalive's ping")
231 }
232}
233
234impl std::error::Error for PeerVanished {}
235
236#[derive(Debug, Default)]
246struct Keepalive {
247 sent: std::sync::Mutex<Option<Instant>>,
253}
254
255#[derive(Debug, PartialEq, Eq)]
257enum Beat {
258 Ping,
260 Wait,
262 Reap,
264}
265
266impl Keepalive {
267 fn sent(&self) -> Option<Instant> {
269 *self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
270 }
271
272 fn heard(&self) {
274 *self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = None;
275 }
276
277 fn beat(&self, now: Instant, dead: Duration) -> Beat {
281 match self.sent() {
282 Some(at) if now.duration_since(at) >= dead => Beat::Reap,
283 Some(_) => Beat::Wait,
284 None => Beat::Ping,
285 }
286 }
287
288 async fn ping<S>(&self, sink: &mut S) -> anyhow::Result<()>
295 where
296 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
297 {
298 write_within(WRITE_BOUND, async {
299 sink.send(Message::Ping(tokio_tungstenite::tungstenite::Bytes::new()))
300 .await
301 .map_err(anyhow::Error::from)
302 })
303 .await?;
304 *self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Instant::now());
305 Ok(())
306 }
307}
308
309struct AttachedClient(Arc<ViewSurface>);
314
315impl AttachedClient {
316 fn new(surface: &Arc<ViewSurface>) -> Self {
317 surface.client_attached();
318 Self(Arc::clone(surface))
319 }
320}
321
322impl Drop for AttachedClient {
323 fn drop(&mut self) {
324 self.0.client_detached();
325 }
326}
327
328struct Viewing {
331 key: String,
332 id: u16,
333 tab: u16,
336 frames: tokio::sync::watch::Receiver<Option<BrowserFrame>>,
337 view: tokio::sync::watch::Receiver<BrowserViewState>,
338 nav: tokio::sync::watch::Receiver<Option<forge_primitives::browser::BrowserNavState>>,
339 failures: tokio::sync::watch::Receiver<Option<String>>,
340 files: tokio::sync::watch::Receiver<Option<forge_primitives::browser::BrowserFileAsk>>,
341 downloads: tokio::sync::broadcast::Receiver<forge_primitives::browser::BrowserDownload>,
342 audio: tokio::sync::broadcast::Receiver<forge_primitives::browser::BrowserAudio>,
343 last_frame: Option<u32>,
346}
347
348impl Viewing {
349 fn new(attachment: BrowserAttachment) -> Self {
350 Self {
351 key: attachment.key,
352 id: attachment.id,
353 tab: attachment.tab,
354 frames: attachment.frames,
355 view: attachment.view,
356 nav: attachment.nav,
357 failures: attachment.failures,
358 files: attachment.files,
359 downloads: attachment.downloads,
360 audio: attachment.audio,
361 last_frame: None,
362 }
363 }
364}
365
366#[derive(Default)]
372struct BrowserViewing {
373 viewing: Option<Viewing>,
374 tabs: Option<tokio::sync::watch::Receiver<BrowserTabList>>,
375}
376
377struct TerminalViewing {
379 seat: forge_primitives::SessionSlot,
380 id: u16,
381 entry: std::sync::Arc<forge_workspace::terminal::TerminalEntry>,
382 output: tokio::sync::broadcast::Receiver<forge_workspace::terminal::OutputChunk>,
383 control: tokio::sync::watch::Receiver<Option<u64>>,
386 viewer: u64,
389 paused: bool,
392 last_seq: u32,
395}
396
397#[derive(Default)]
399struct TerminalViewingState {
400 viewing: Option<TerminalViewing>,
401 role: Option<(std::sync::Arc<forge_workspace::terminal::TerminalEntry>, u64)>,
405}
406
407impl Drop for TerminalViewingState {
408 fn drop(&mut self) {
409 if let Some((entry, viewer)) = self.role.take() {
410 entry.release_control(viewer);
411 }
412 }
413}
414
415#[derive(Default)]
421struct Viewers {
422 browser: BrowserViewing,
423 terminal: TerminalViewingState,
424}
425
426impl Drop for TerminalViewing {
427 fn drop(&mut self) {
428 self.entry.attached.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
433 }
434}
435
436async fn send_terminal_view<S>(
439 socket: &mut ClientSocket<S>,
440 viewing: &TerminalViewing,
441 repaint: &[u8],
442) -> anyhow::Result<()>
443where
444 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
445{
446 send(
447 socket,
448 ServerMessage::TerminalView {
449 seat: viewing.seat.clone(),
450 id: viewing.id,
451 cols: viewing.entry.terminal.cols(),
452 rows: viewing.entry.terminal.rows(),
453 seq: viewing.last_seq,
454 screen: String::from_utf8_lossy(repaint).into_owned(),
455 viewers: u32::try_from(
456 viewing.entry.attached.load(std::sync::atomic::Ordering::Relaxed),
457 )
458 .unwrap_or(u32::MAX),
459 viewer: viewing.viewer,
460 controller: viewing.entry.controller(),
461 },
462 )
463 .await
464}
465
466enum TerminalEvent {
468 Chunk(forge_workspace::terminal::OutputChunk),
469 Control(Option<u64>),
471 Resync,
474 Gone,
476}
477
478async fn next_terminal_event(viewing: &mut Option<TerminalViewing>) -> Option<TerminalEvent> {
480 let Some(viewing) = viewing.as_mut() else { return std::future::pending().await };
481 let mut exit = viewing.entry.terminal.exit_stream();
485 if !viewing.entry.terminal.is_alive() {
486 return Some(TerminalEvent::Gone);
487 }
488 loop {
489 let chunk = tokio::select! {
490 chunk = viewing.output.recv() => chunk,
491 _ = exit.changed() => return Some(TerminalEvent::Gone),
492 changed = viewing.control.changed() => {
493 if changed.is_ok() {
494 return Some(TerminalEvent::Control(*viewing.control.borrow()));
495 }
496 continue;
497 }
498 };
499 match chunk {
500 Ok(chunk) => {
501 if viewing.paused {
502 viewing.last_seq = chunk.seq;
503 continue;
504 }
505 viewing.last_seq = chunk.seq;
506 return Some(TerminalEvent::Chunk(chunk));
507 }
508 Err(tokio::sync::broadcast::error::RecvError::Lagged(lost)) => {
509 tracing::debug!(
510 event_name = "terminal_viewer_lagged",
511 id = viewing.id,
512 lost,
513 "a terminal viewer fell behind and is being resynced"
514 );
515 return Some(TerminalEvent::Resync);
516 }
517 Err(tokio::sync::broadcast::error::RecvError::Closed) => {
518 return Some(TerminalEvent::Gone);
519 }
520 }
521 }
522}
523
524enum ViewEvent {
526 Frame(BrowserFrame),
527 State(BrowserViewState),
528 Nav(forge_primitives::browser::BrowserNavState),
529 Failed(String),
530 File(forge_primitives::browser::BrowserFileAsk),
531 Download {
532 download: forge_primitives::browser::BrowserDownload,
533 dropped: bool,
536 lost: Option<u64>,
539 },
540 Audio(forge_primitives::browser::BrowserAudio),
541}
542
543const _: () = assert!(
548 forge_primitives::browser::DOWNLOAD_BACKLOG == 4,
549 "the drop notice says 'four' finished downloads; change the words with the backlog"
550);
551
552async fn next_view_event(viewing: &mut Option<Viewing>) -> ViewEvent {
559 let Some(viewing) = viewing.as_mut() else { return std::future::pending().await };
560 loop {
561 let ready = tokio::select! {
566 changed = viewing.frames.changed() => changed.map(|()| 0_u8).map_err(|_| 10_u8),
567 changed = viewing.view.changed() => changed.map(|()| 1_u8).map_err(|_| 11_u8),
568 changed = viewing.failures.changed() => changed.map(|()| 2_u8).map_err(|_| 12_u8),
569 changed = viewing.nav.changed() => changed.map(|()| 3_u8).map_err(|_| 13_u8),
570 changed = viewing.files.changed() => changed.map(|()| 4_u8).map_err(|_| 14_u8),
571 download = viewing.downloads.recv() => match download {
572 Ok(download) => {
573 return ViewEvent::Download { download, dropped: false, lost: None };
574 }
575 Err(tokio::sync::broadcast::error::RecvError::Lagged(lost)) => {
576 tracing::warn!(
577 event_name = "browser_download_lagged",
578 profile = %viewing.key,
579 profile_id = viewing.id,
580 tab = viewing.tab,
581 lost,
582 "a viewer fell behind a finished download and lost it"
583 );
584 let why = if lost == 1 {
597 "a finished download was dropped before this viewer read it: the \
598 profile's backlog holds at most four finished downloads at a time"
599 .to_owned()
600 } else {
601 format!(
602 "{lost} finished downloads were dropped before this viewer read \
603 them: the profile's backlog holds at most four finished downloads \
604 at a time"
605 )
606 };
607 return ViewEvent::Download {
608 download: forge_primitives::browser::BrowserDownload {
609 tab: None,
610 name: "download".to_owned(),
611 bytes: Vec::new(),
612 error: Some(why),
613 },
614 dropped: true,
615 lost: Some(lost),
616 };
617 }
618 Err(tokio::sync::broadcast::error::RecvError::Closed) => {
619 std::future::pending().await
620 }
621 },
622 packet = viewing.audio.recv() => match packet {
623 Ok(packet) if packet.tab == viewing.tab => return ViewEvent::Audio(packet),
627 Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
631 Err(tokio::sync::broadcast::error::RecvError::Closed) => {
632 std::future::pending().await
633 }
634 },
635 };
636 match ready {
637 Ok(0) => {
638 let frame = viewing.frames.borrow_and_update().clone();
639 if let Some(frame) = frame
643 && frame.tab == viewing.tab
644 && viewing.last_frame != Some(frame.seq)
645 {
646 viewing.last_frame = Some(frame.seq);
647 return ViewEvent::Frame(frame);
648 }
649 }
650 Ok(1) => return ViewEvent::State(*viewing.view.borrow_and_update()),
651 Ok(2) => {
652 if let Some(reason) = viewing.failures.borrow_and_update().clone() {
653 return ViewEvent::Failed(reason);
654 }
655 }
656 Ok(3) => {
657 if let Some(nav) = *viewing.nav.borrow_and_update() {
658 return ViewEvent::Nav(nav);
659 }
660 }
661 Ok(4) => {
662 if let Some(ask) = *viewing.files.borrow_and_update() {
663 return ViewEvent::File(ask);
664 }
665 }
666 Ok(_) => {}
667 Err(10 | 11 | 13) => {
675 if let Some(reason) = viewing.failures.borrow().clone() {
676 return ViewEvent::Failed(reason);
677 }
678 return ViewEvent::Failed(format!("tab {} was closed", viewing.tab));
679 }
680 Err(_) => {
681 if let Some(reason) = viewing.failures.borrow().clone() {
682 return ViewEvent::Failed(reason);
683 }
684 return ViewEvent::Failed("the browser was closed".to_owned());
685 }
686 }
687 }
688}
689
690async fn next_tab_change(
694 tabs: &mut Option<tokio::sync::watch::Receiver<BrowserTabList>>,
695) -> Option<BrowserTabList> {
696 let Some(tabs) = tabs.as_mut() else { return std::future::pending().await };
697 match tabs.changed().await {
698 Ok(()) => Some(tabs.borrow_and_update().clone()),
699 Err(_) => std::future::pending().await,
702 }
703}
704
705const MAX_CLIENT_MESSAGE_BYTES: usize = 16 * 1024 * 1024 + 10;
713
714pub async fn upgrade(
724 ws: WebSocketUpgrade,
725 State(state): State<Arc<TransportState>>,
726 ConnectInfo(peer): ConnectInfo<std::net::SocketAddr>,
727 headers: HeaderMap,
728 uri: axum::http::Uri,
729) -> Response {
730 let mut origin: Option<&str> = None;
733 let mut malformed = false;
734 for value in headers.get_all(header::ORIGIN) {
735 match (value.to_str(), origin) {
736 (Ok(text), None) => origin = Some(text),
737 _ => malformed = true,
738 }
739 }
740 if malformed || !origin_gate::allows(origin, &state.surface.server_config()) {
741 let refused_origin =
744 if malformed { "<malformed>" } else { origin.unwrap_or("<malformed>") };
745 tracing::warn!(
746 target: "forge_server::transport",
747 event_name = "socket_origin_refused",
748 origin = refused_origin,
749 peer = %peer,
750 "refused a socket upgrade from an origin forge does not serve",
751 );
752 return (StatusCode::FORBIDDEN, "origin refused").into_response();
753 }
754 let deflate = uri.query().is_some_and(|query| query.split('&').any(|pair| pair == "deflate=1"));
760 let limit = MAX_CLIENT_MESSAGE_BYTES;
761 ws.max_frame_size(limit)
762 .max_message_size(limit)
763 .on_upgrade(move |socket| greet(socket, state, peer, deflate))
764}
765
766fn greeting_for(client: &forge_primitives::ClientConfig, state: &TransportState) -> ServerMessage {
770 ServerMessage::Greeting {
771 version: PROTOCOL_VERSION,
772 forge_version: crate::FORGE_VERSION.to_owned(),
773 forge_version_short: crate::FORGE_VERSION_SHORT.to_owned(),
774 settings: ClientSettings::new(client, state.surface.dictate_axes()),
775 }
776}
777
778#[derive(Clone, Copy)]
781struct Caller {
782 id: u64,
783 peer: std::net::SocketAddr,
784}
785
786#[derive(Clone, Copy)]
790struct Arrival {
791 caller: Caller,
792 at: Instant,
793}
794
795async fn greet<S>(socket: S, state: Arc<TransportState>, peer: std::net::SocketAddr, deflate: bool)
802where
803 S: futures_util::Stream<Item = Result<Message, axum::Error>>
804 + futures_util::Sink<Message, Error = axum::Error>
805 + Unpin
806 + Send,
807{
808 let mut socket = ClientSocket::new(socket, deflate);
809 let client = state.surface.client_config();
810 let greeting = greeting_for(&client, &state);
811 let Ok(text) = serde_json::to_string(&greeting) else {
812 tracing::error!(
813 target: "forge_server::transport",
814 event_name = "greeting_unencodable",
815 "the greeting could not be encoded, so this connection has nothing to say",
816 );
817 return;
818 };
819 if write_within(WRITE_BOUND, async {
820 socket.send(Message::Text(text.into())).await.map_err(anyhow::Error::from)
821 })
822 .await
823 .is_err()
824 {
825 return;
826 }
827 let settings = state.settings.subscribe();
832 let sent_settings = state.settings.borrow().clone();
833 if let Some(text) = &sent_settings
834 && send_text(&mut socket, text).await.is_err()
835 {
836 return;
837 }
838 let progress = Arc::new(Progress::new());
842 let mut socket = socket.watched(Arc::clone(&progress));
843 if let Err(error) =
844 supervise(&mut socket, &state, peer, settings, sent_settings, &progress).await
845 {
846 tracing::debug!(
847 target: "forge_server::transport",
848 event_name = "connection_ended",
849 %error,
850 "the connection ended on an error rather than on the client closing",
851 );
852 }
853}
854
855struct ConnectionLife {
861 me: u64,
865 watching: Watching,
866 dictate: Streaming,
874}
875
876impl ConnectionLife {
877 fn new() -> Self {
878 Self {
879 me: mint_connection_id(),
880 watching: Watching::default(),
881 dictate: Streaming::default(),
882 }
883 }
884}
885
886async fn supervise<S>(
896 socket: &mut ClientSocket<S>,
897 state: &Arc<TransportState>,
898 peer: std::net::SocketAddr,
899 settings: tokio::sync::watch::Receiver<Option<Arc<str>>>,
900 sent_settings: Option<Arc<str>>,
901 progress: &Progress,
902) -> anyhow::Result<()>
903where
904 S: futures_util::Stream<Item = Result<Message, axum::Error>>
905 + futures_util::Sink<Message, Error = axum::Error>
906 + Unpin
907 + Send,
908{
909 let mut life = ConnectionLife::new();
910 let me = life.me;
911 let outcome = {
915 let connection = drive(socket, state, peer, settings, sent_settings, &mut life);
916 tokio::pin!(connection);
917 let mut sample = tokio::time::interval(state.heartbeat.every);
918 sample.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
919 sample.tick().await;
921 loop {
922 tokio::select! {
923 biased;
927 outcome = &mut connection => break outcome,
928 _ = sample.tick() => {
929 let silent = progress.silent_for(tokio::time::Instant::now());
930 if silent < state.heartbeat.wedge {
931 continue;
932 }
933 let slot = progress.showing_seat();
939 tracing::warn!(
940 target: "forge_server::transport",
941 event_name = "connection_wedged",
942 client = me,
943 peer = %peer,
944 slot = ?slot,
945 silent_ms = silent.as_secs_f64() * 1000.0,
946 bound_seconds = state.heartbeat.wedge.as_secs(),
947 "no write completed on this connection past its bound, so it is ended: \
948 something it awaited never answered",
949 );
950 break Err(anyhow::Error::new(WedgeStalled {
951 bound: state.heartbeat.wedge,
952 silent,
953 }));
954 }
955 }
956 }
957 };
958 if let Err(error) = &outcome
959 && let Some(stalled) = error.downcast_ref::<WriteStalled>()
960 {
961 tracing::debug!(
962 target: "forge_server::transport",
963 event_name = "connection_write_stalled",
964 client = me,
965 bound_seconds = stalled.bound.as_secs(),
966 waited_ms = stalled.waited.as_secs_f64() * 1000.0,
967 "a frame write outlived its bound, so the connection is ended: the peer stopped \
968 reading",
969 );
970 }
971 finish(state, &mut life, &outcome).await;
972 outcome
973}
974
975fn ending(outcome: &anyhow::Result<()>) -> &'static str {
979 match outcome {
980 Ok(()) => "closed",
981 Err(error) if error.downcast_ref::<WedgeStalled>().is_some() => "wedged",
982 Err(error) if error.downcast_ref::<PeerVanished>().is_some() => "vanished",
983 Err(_) => "error",
984 }
985}
986
987async fn drive<S>(
996 socket: &mut ClientSocket<S>,
997 state: &Arc<TransportState>,
998 peer: std::net::SocketAddr,
999 settings: tokio::sync::watch::Receiver<Option<Arc<str>>>,
1000 sent_settings: Option<Arc<str>>,
1001 life: &mut ConnectionLife,
1002) -> anyhow::Result<()>
1003where
1004 S: futures_util::Stream<Item = Result<Message, axum::Error>>
1005 + futures_util::Sink<Message, Error = axum::Error>
1006 + Unpin
1007 + Send,
1008{
1009 let me = life.me;
1015 let caller = Caller { id: me, peer };
1016 let _attached = AttachedClient::new(&state.surface);
1019 let mut holds = Holds::new(&state.surface);
1025 let mut updates: Option<(mpsc::UnboundedReceiver<SeatUpdate>, bool)> = None;
1032 let mut viewers = Viewers::default();
1035 let (reply_tx, mut answers) = mpsc::unbounded_channel();
1039 let mut reads =
1045 Reads { one: None, waiting: std::collections::VecDeque::new(), reply: reply_tx };
1046
1047 run_connection(
1048 socket,
1049 state,
1050 &mut life.watching,
1051 &mut holds,
1052 &mut updates,
1053 &mut life.dictate,
1054 &mut viewers,
1055 AnswerLanes { reads: &mut reads, answers: &mut answers },
1056 SettingsPush { receiver: settings, sent: sent_settings },
1057 caller,
1058 )
1059 .await
1060}
1061
1062async fn finish(
1069 state: &Arc<TransportState>,
1070 life: &mut ConnectionLife,
1071 outcome: &anyhow::Result<()>,
1072) {
1073 let me = life.me;
1074 let ended = ending(outcome);
1075 let dictate = &mut life.dictate;
1076 let watching = &mut life.watching;
1077 for claim in &dictate.seats {
1083 if state.surface.dictate_close(&claim.seat, me) {
1084 tracing::debug!(
1085 target: "forge_server::transport",
1086 event_name = "dictate_take_dropped",
1087 ending = ended,
1088 slot = %claim.seat.display(),
1089 "the connection that was streaming a take went away; the take was dropped",
1090 );
1091 }
1092 }
1093 let released = state.surface.voice_release(me);
1097 if !released.is_empty() {
1098 tracing::debug!(
1099 target: "forge_server::transport",
1100 event_name = "voice_listener_released",
1101 ending = ended,
1102 seats = released.len(),
1103 "the connection that was listening went away; its seats drop it",
1104 );
1105 }
1106 if dictate.read_aloud {
1110 let stopped = state
1111 .surface
1112 .dispatch(Command::DictateReadAloudStop { keep: false, initiator: Some(me) });
1113 tracing::debug!(
1114 target: "forge_server::transport",
1115 event_name = "read_aloud_recording_dropped",
1116 ending = ended,
1117 stopped = stopped.is_ok(),
1118 "the connection that was recording the read-aloud set went away",
1119 );
1120 }
1121
1122 if watching.showing_attached
1128 && let Some(slot) = watching.showing.as_ref()
1129 {
1130 Live::lock(&state.live).detach(slot);
1131 }
1132 if let Some(host) = state.browser_host.as_ref()
1140 && tokio::time::timeout(WRITE_BOUND, host.detach(me)).await.is_err()
1141 {
1142 let slot = watching.showing.as_ref().map_or(String::new(), |seat| seat.display().clone());
1145 tracing::warn!(
1146 target: "forge_server::transport",
1147 event_name = "browser_detach_stalled",
1148 client = me,
1149 ending = ended,
1150 slot = %slot,
1151 bound_seconds = WRITE_BOUND.as_secs(),
1152 "the host did not take the viewer's detach within the bound, so the profile keeps a \
1153 viewer that is gone",
1154 );
1155 }
1156 }
1162
1163#[derive(Default)]
1170struct Streaming {
1171 seats: Vec<Claim>,
1172 read_aloud: bool,
1173}
1174
1175struct Claim {
1183 seat: SessionSlot,
1184 generation: u64,
1185}
1186
1187impl Streaming {
1188 fn any(&self) -> bool {
1190 !self.seats.is_empty() || self.read_aloud
1191 }
1192
1193 fn claim(&mut self, seat: &SessionSlot) {
1200 match self.seats.iter_mut().find(|claim| &claim.seat == seat) {
1201 Some(claim) => claim.generation = 0,
1202 None => self.seats.push(Claim { seat: seat.clone(), generation: 0 }),
1203 }
1204 }
1205
1206 fn stamp(&mut self, seat: &SessionSlot, generation: u64) {
1208 if let Some(claim) = self.seats.iter_mut().find(|claim| &claim.seat == seat) {
1209 claim.generation = generation;
1210 }
1211 }
1212
1213 fn retire(&mut self, seat: &SessionSlot, outcome: &DictateOutcome, generation: u64) {
1217 if matches!(outcome, DictateOutcome::Refused { .. }) {
1218 return;
1219 }
1220 self.seats.retain(|claim| &claim.seat != seat || claim.generation != generation);
1221 }
1222}
1223
1224enum Answer {
1226 Reply(Box<ServerMessage>),
1228 Read(Box<ReadAnswer>),
1230}
1231
1232struct Reads {
1235 one: Option<Reading>,
1236 waiting: std::collections::VecDeque<Queued>,
1239 reply: mpsc::UnboundedSender<Answer>,
1240}
1241
1242enum Queued {
1249 Ask(Instant, ClientMessage),
1250 Withdrawn(Subject),
1251}
1252
1253const MAX_HELD_READ_FRAMES: usize = 1024;
1257
1258struct ReadAnswer {
1260 subject: Subject,
1261 seat: Option<SessionSlot>,
1262 peer: std::net::SocketAddr,
1263 open_at: Instant,
1266 queue_ms: f64,
1268 prelude_ms: f64,
1269 queued_ms: f64,
1270 hold_ms: f64,
1271 outcome: anyhow::Result<(serde_json::Value, OpenTimings)>,
1272}
1273
1274struct Reading {
1283 subject: Subject,
1284 frames: Vec<SeatUpdate>,
1285 _task: AbortOnDrop,
1288}
1289
1290struct ReadAnswerGuard {
1295 reply: mpsc::UnboundedSender<Answer>,
1296 subject: Subject,
1297 seat: Option<SessionSlot>,
1298 peer: std::net::SocketAddr,
1299 open_at: Instant,
1300 queue_ms: f64,
1301 prelude_ms: f64,
1302 queued_ms: f64,
1303 hold_ms: f64,
1304 answered: bool,
1305}
1306
1307impl ReadAnswerGuard {
1308 fn answer(&mut self, outcome: anyhow::Result<(serde_json::Value, OpenTimings)>) {
1309 self.answered = true;
1310 let _ = self.reply.send(Answer::Read(Box::new(ReadAnswer {
1311 subject: self.subject.clone(),
1312 seat: self.seat.clone(),
1313 peer: self.peer,
1314 open_at: self.open_at,
1315 queue_ms: self.queue_ms,
1316 prelude_ms: self.prelude_ms,
1317 queued_ms: self.queued_ms,
1318 hold_ms: self.hold_ms,
1319 outcome,
1320 })));
1321 }
1322}
1323
1324impl Drop for ReadAnswerGuard {
1325 fn drop(&mut self) {
1326 if self.answered || !std::thread::panicking() {
1329 return;
1330 }
1331 self.answer(Err(anyhow::anyhow!("the read ended without an answer")));
1332 }
1333}
1334
1335struct AbortOnDrop(tokio::task::JoinHandle<()>);
1338
1339impl Drop for AbortOnDrop {
1340 fn drop(&mut self) {
1341 self.0.abort();
1342 }
1343}
1344
1345fn spawn_read(
1350 mut answer: ReadAnswerGuard,
1351 encode: impl std::future::Future<Output = anyhow::Result<(serde_json::Value, OpenTimings)>>
1352 + Send
1353 + 'static,
1354) -> AbortOnDrop {
1355 AbortOnDrop(tokio::spawn(async move {
1356 let outcome = encode.await;
1357 answer.answer(outcome);
1358 }))
1359}
1360
1361struct AnswerLanes<'a> {
1366 reads: &'a mut Reads,
1367 answers: &'a mut mpsc::UnboundedReceiver<Answer>,
1368}
1369
1370struct SettingsPush {
1373 receiver: tokio::sync::watch::Receiver<Option<Arc<str>>>,
1374 sent: Option<Arc<str>>,
1375}
1376
1377async fn run_connection<S>(
1378 socket: &mut ClientSocket<S>,
1379 state: &Arc<TransportState>,
1380 watching: &mut Watching,
1381 holds: &mut Holds<'_>,
1382 updates: &mut Option<(mpsc::UnboundedReceiver<SeatUpdate>, bool)>,
1383 dictate: &mut Streaming,
1384 viewers: &mut Viewers,
1385 lanes: AnswerLanes<'_>,
1386 settings_push: SettingsPush,
1387 caller: Caller,
1388) -> anyhow::Result<()>
1389where
1390 S: futures_util::Stream<Item = Result<Message, axum::Error>>
1391 + futures_util::Sink<Message, Error = axum::Error>
1392 + Unpin
1393 + Send,
1394{
1395 let AnswerLanes { reads, answers } = lanes;
1396 let SettingsPush { receiver: settings, sent: sent_settings } = settings_push;
1397 let me = caller.id;
1398 let mut held = Batch::default();
1399 let mut settings = settings;
1400 let mut settings_open = true;
1401 let mut last_settings = sent_settings;
1404 let cap = state.surface.tool_record_cap_bytes();
1407 let beat_every = state.heartbeat.every;
1412 let beat_dead = state.heartbeat.dead;
1413 let mut beat = tokio::time::interval(beat_every);
1414 beat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
1415 beat.tick().await;
1416 let keepalive = Keepalive::default();
1417 let mut vanished = false;
1420 loop {
1421 let due = held.due();
1422 tokio::select! {
1423 biased;
1429 msg = socket.next() => {
1430 let Some(msg) = msg else { break }; let msg = msg?;
1432 keepalive.heard();
1433 batch::flush(socket, held.take()).await?;
1437 let arrival = Arrival { caller, at: Instant::now() };
1438 handle_client(
1439 socket, state, watching, holds, updates, dictate, viewers, reads, arrival, msg,
1440 )
1441 .await?;
1442 while reads.one.is_none() {
1447 let Some(next) = reads.waiting.pop_front() else { break };
1448 match next {
1449 Queued::Ask(at, message) => {
1450 handle_message(
1451 socket, state, watching, holds, updates, dictate, viewers, reads,
1452 Arrival { caller, at }, message,
1453 )
1454 .await?;
1455 }
1456 Queued::Withdrawn(what) => {
1459 send(socket, released_subscribe(&what)).await?;
1460 }
1461 }
1462 }
1463 }
1464 answer = answers.recv() => {
1469 let Some(answer) = answer else { break };
1470 batch::flush(socket, held.take()).await?;
1471 match answer {
1472 Answer::Reply(message) => send(socket, *message).await?,
1473 Answer::Read(read) => {
1474 finish_read(socket, state, watching, holds, &mut reads.one, &mut held, *read)
1475 .await?;
1476 while reads.one.is_none() {
1479 let Some(next) = reads.waiting.pop_front() else { break };
1480 match next {
1481 Queued::Ask(at, message) => {
1482 handle_message(
1483 socket, state, watching, holds, updates, dictate, viewers,
1484 reads, Arrival { caller, at }, message,
1485 )
1486 .await?;
1487 }
1488 Queued::Withdrawn(what) => {
1489 send(socket, released_subscribe(&what)).await?;
1490 }
1491 }
1492 }
1493 }
1494 }
1495 }
1496 settings_changed = settings.changed(), if settings_open => {
1502 if settings_changed.is_err() {
1503 settings_open = false;
1504 } else {
1505 let current = settings.borrow_and_update().clone();
1506 if current != last_settings {
1507 batch::flush(socket, held.take()).await?;
1508 if let Some(text) = ¤t {
1509 send_text(socket, text).await?;
1510 }
1511 last_settings = current;
1512 }
1513 }
1514 }
1515 heard = next_update(updates) => {
1519 let Some(frame) = heard else { break };
1520 let update = frame.update;
1521 if let SessionUpdate::DictateStarted {
1531 key,
1532 generation,
1533 initiator: Some(id),
1534 ..
1535 } = &*update
1536 && *id == me
1537 {
1538 dictate.stamp(key, *generation);
1539 }
1540 if let SessionUpdate::DictateEnded {
1541 key,
1542 initiator: Some(id),
1543 outcome,
1544 generation,
1545 ..
1546 } = &*update
1547 && *id == me
1548 {
1549 dictate.retire(key, outcome, *generation);
1550 }
1551 let held_for_read = reads.one.as_ref().is_some_and(|read| {
1563 ours_to_hear(&update, me) && read.subject.covers(&update)
1564 });
1565 if held_for_read
1566 && let Some(read) = reads.one.as_mut()
1567 {
1568 if read.frames.len() == MAX_HELD_READ_FRAMES {
1571 read.frames.remove(0);
1572 }
1573 read.frames.push(SeatUpdate {
1574 seq: frame.seq,
1575 update: Arc::new(for_a_client((*update).clone(), cap)),
1576 });
1577 }
1578 if forwards(&update, &watching.subjects, me) {
1581 let fatal = matches!(&*update, SessionUpdate::FatalError { .. });
1587 held.push(
1588 Instant::now(),
1589 SeatUpdate {
1590 seq: frame.seq,
1591 update: Arc::new(for_a_client(Arc::unwrap_or_clone(update), cap)),
1592 },
1593 );
1594 if fatal {
1595 batch::flush(socket, held.take()).await?;
1596 }
1597 }
1598 }
1599 event = next_view_event(&mut viewers.browser.viewing) => {
1602 batch::flush(socket, held.take()).await?;
1605 match event {
1606 ViewEvent::Frame(frame) => {
1607 let mut transport = super::frame::SocketFrames { socket: &mut *socket };
1611 transport.deliver(&frame).await?;
1612 }
1613 ViewEvent::State(state) => {
1614 let Some(viewing) = viewers.browser.viewing.as_ref() else { continue };
1615 send(
1616 socket,
1617 ServerMessage::BrowserView {
1618 id: viewing.id,
1619 profile: viewing.key.clone(),
1620 tab: viewing.tab,
1621 controller: state.controller,
1622 viewers: state.viewers,
1623 viewer: me,
1624 },
1625 )
1626 .await?;
1627 }
1628 ViewEvent::Nav(nav) => {
1631 let Some(viewing) = viewers.browser.viewing.as_ref() else { continue };
1632 send(
1633 socket,
1634 ServerMessage::BrowserNavState {
1635 id: viewing.id,
1636 tab: viewing.tab,
1637 can_back: nav.can_back,
1638 can_forward: nav.can_forward,
1639 },
1640 )
1641 .await?;
1642 }
1643 ViewEvent::File(ask) => {
1655 let Some(viewing) = viewers.browser.viewing.as_ref() else { continue };
1656 if ask.tab != viewing.tab {
1659 continue;
1660 }
1661 let controller = viewing.view.borrow().controller;
1662 if controller.is_some_and(|holder| holder != me) {
1663 continue;
1664 }
1665 if controller.is_none() {
1666 if viewing.files.borrow().is_some()
1670 && let Some(host) = state.browser_host.as_ref()
1671 && let Err(why) = host
1672 .answer_files(&viewing.key, viewing.tab, ask.id, Vec::new())
1673 .await
1674 {
1675 tracing::warn!(
1676 "a file chooser could not be cancelled without a controller \
1677 (event_name browser_file_cancel_failed, profile {}, id {}): \
1678 {why}",
1679 viewing.key,
1680 ask.id,
1681 );
1682 }
1683 continue;
1684 }
1685 send(
1686 socket,
1687 ServerMessage::BrowserFileDialog {
1688 id: ask.id,
1689 tab: ask.tab,
1690 multiple: ask.multiple,
1691 },
1692 )
1693 .await?;
1694 }
1695 ViewEvent::Download { download, dropped, lost } => {
1699 let Some(viewing) = viewers.browser.viewing.as_ref() else { continue };
1700 if download.tab.is_some_and(|tab| tab != viewing.tab) {
1705 continue;
1706 }
1707 if viewing.view.borrow().controller != Some(me) {
1708 continue;
1709 }
1710 let data = download.error.is_none().then(|| {
1711 base64::engine::general_purpose::STANDARD.encode(&download.bytes)
1712 });
1713 send(
1714 socket,
1715 ServerMessage::BrowserDownload {
1716 tab: download.tab,
1717 name: download.name,
1718 data,
1719 error: download.error,
1720 dropped,
1721 lost,
1722 },
1723 )
1724 .await?;
1725 }
1726 ViewEvent::Audio(packet) => {
1730 let Some(viewing) = viewers.browser.viewing.as_ref() else { continue };
1731 if viewing.view.borrow().controller != Some(me) {
1732 continue;
1733 }
1734 let mut transport = super::frame::SocketFrames { socket: &mut *socket };
1735 transport.deliver_audio(&packet).await?;
1736 }
1737 ViewEvent::Failed(reason) => {
1742 let Some(viewed) = viewers.browser.viewing.as_ref() else { continue };
1743 let (id, tab) = (viewed.id, viewed.tab);
1744 viewers.browser.viewing = None;
1745 if let Some(host) = state.browser_host.as_ref() {
1750 host.detach(me).await;
1751 }
1752 send(socket, ServerMessage::BrowserSurfaceFailed { id, tab, reason })
1753 .await?;
1754 }
1755 }
1756 }
1757 changed = next_tab_change(&mut viewers.browser.tabs) => {
1761 let Some(list) = changed else { continue };
1762 batch::flush(socket, held.take()).await?;
1763 send(
1764 socket,
1765 ServerMessage::BrowserTabs {
1766 id: list.id,
1767 profile: list.profile,
1768 up: list.up,
1769 connected: list.connected,
1770 closing_at_ms: list.closing_at_ms,
1771 rows: list.rows,
1772 },
1773 )
1774 .await?;
1775 }
1776 event = next_terminal_event(&mut viewers.terminal.viewing) => {
1780 batch::flush(socket, held.take()).await?;
1781 let Some(event) = event else { continue };
1782 match event {
1783 TerminalEvent::Chunk(chunk) => {
1784 let Some(viewing) = viewers.terminal.viewing.as_ref() else { continue };
1785 let mut transport = super::frame::SocketFrames { socket: &mut *socket };
1786 transport.deliver_terminal(viewing.id, chunk.seq, &chunk.bytes).await?;
1787 }
1788 TerminalEvent::Control(controller) => {
1789 let Some(viewing) = viewers.terminal.viewing.as_ref() else { continue };
1790 let held = controller == Some(viewing.viewer);
1794 let entry = viewing.entry.clone();
1795 viewers.terminal.role = if held { Some((entry, viewing.viewer)) } else { None };
1796 send(
1797 socket,
1798 ServerMessage::TerminalControl {
1799 seat: viewing.seat.clone(),
1800 id: viewing.id,
1801 controller,
1802 },
1803 )
1804 .await?;
1805 }
1806 TerminalEvent::Resync => {
1807 if let Some(viewing) = viewers.terminal.viewing.as_mut() {
1808 let (repaint, seq) = viewing.entry.terminal.repaint_and_seq();
1809 viewing.last_seq = seq.saturating_sub(1);
1810 send_terminal_view(socket, viewing, &repaint).await?;
1811 }
1812 }
1813 TerminalEvent::Gone => {
1814 viewers.terminal.viewing = None;
1818 }
1819 }
1820 }
1821 _ = beat.tick() => {
1832 match keepalive.beat(Instant::now(), beat_dead) {
1833 Beat::Wait => {}
1834 Beat::Ping => keepalive.ping(socket).await?,
1839 Beat::Reap => {
1840 let waited =
1841 keepalive.sent().map(|at| at.elapsed()).unwrap_or_default();
1842 tracing::debug!(
1843 target: "forge_server::transport",
1844 event_name = "connection_reaped",
1845 client = me,
1846 waited_ms = waited.as_secs_f64() * 1000.0,
1847 "the peer never answered the keepalive's ping, so its connection is ended",
1848 );
1849 vanished = true;
1853 break;
1854 }
1855 }
1856 }
1857 () = tokio::time::sleep_until(due.unwrap_or_else(Instant::now).into()),
1860 if due.is_some() =>
1861 {
1862 batch::flush(socket, held.take()).await?;
1863 }
1864 }
1865 }
1866 if vanished {
1867 return Err(anyhow::Error::new(PeerVanished));
1868 }
1869 Ok(())
1870}
1871
1872fn open_stream(
1894 state: &TransportState,
1895 updates: &mut Option<(mpsc::UnboundedReceiver<SeatUpdate>, bool)>,
1896 answering: bool,
1897) -> Vec<SeatUpdate> {
1898 if updates.as_ref().is_some_and(|(_, held)| *held >= answering) {
1899 return Vec::new();
1900 }
1901 let queued = updates.take().map_or_else(Vec::new, |(mut receiver, _)| {
1902 let mut queued = Vec::new();
1903 while let Ok(frame) = receiver.try_recv() {
1904 queued.push(frame);
1905 }
1906 queued
1907 });
1908 *updates = Some((state.surface.subscribe_client(answering), answering));
1909 queued
1910}
1911
1912async fn next_update(
1916 updates: &mut Option<(mpsc::UnboundedReceiver<SeatUpdate>, bool)>,
1917) -> Option<SeatUpdate> {
1918 match updates.as_mut() {
1919 Some((updates, _)) => updates.recv().await,
1920 None => std::future::pending().await,
1921 }
1922}
1923
1924fn stamp(command: Command, me: u64) -> Command {
1933 match command {
1934 Command::DictateStream { key, options, surface, .. } => {
1935 Command::DictateStream { key, options, surface, initiator: Some(me) }
1936 }
1937 Command::DictateStop { key, submit, .. } => {
1938 Command::DictateStop { key, submit, initiator: Some(me) }
1939 }
1940 Command::DictateReadAloudStart { .. } => {
1941 Command::DictateReadAloudStart { initiator: Some(me) }
1942 }
1943 Command::DictateReadAloudStop { keep, .. } => {
1944 Command::DictateReadAloudStop { keep, initiator: Some(me) }
1945 }
1946 Command::VoiceMode { key, on, explicit, .. } => {
1949 Command::VoiceMode { key, on, explicit, initiator: Some(me) }
1950 }
1951 Command::RespondBrowserHandOff { key, id, mut answer } => {
1952 answer.client = Some(me);
1953 Command::RespondBrowserHandOff { key, id, answer }
1954 }
1955 Command::RespondTerminalHandOff { key, id, mut answer } => {
1956 answer.client = Some(me);
1957 Command::RespondTerminalHandOff { key, id, answer }
1958 }
1959 other => other,
1960 }
1961}
1962
1963struct OpenReport {
1965 seat: Option<SessionSlot>,
1966 peer: std::net::SocketAddr,
1967 open_at: Instant,
1970 queue_ms: f64,
1972 prelude_ms: f64,
1973 queued_ms: f64,
1974 hold_ms: f64,
1975 timings: OpenTimings,
1976}
1977
1978async fn write_snapshot<S>(
1985 socket: &mut ClientSocket<S>,
1986 state: &Arc<TransportState>,
1987 watching: &mut Watching,
1988 subject: Subject,
1989 data: serde_json::Value,
1990 report: Option<OpenReport>,
1991) -> anyhow::Result<()>
1992where
1993 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
1994{
1995 let seat = report.as_ref().and_then(|report| report.seat.clone());
2003 watching.subjects.push(subject.clone());
2004 let write =
2005 send_frame_timed(socket, ServerMessage::Snapshot { subject, data }, &SystemClock).await?;
2006 if let Some(report) = &report {
2007 log_open(report, &write);
2008 }
2009 if let Some(slot) = &seat
2014 && watching.showing.as_ref() == Some(slot)
2015 && !watching.showing_attached
2016 {
2017 spend_showing(socket, state, watching, slot).await?;
2018 }
2019 Ok(())
2020}
2021
2022fn log_open(report: &OpenReport, write: &FrameWrite) {
2024 let Some(slot) = &report.seat else { return };
2025 tracing::debug!(
2028 event_name = "frame_write",
2029 slot = %slot.display(),
2030 peer = %report.peer,
2031 bytes = write.bytes,
2032 prep_ms = write.prep_ms,
2033 write_ms = write.write_ms,
2034 waited = write.waited,
2035 "the snapshot's write: prep is the record serialised and \
2036 compressed into the bytes the socket takes, write_ms is the \
2037 socket's share after that, and waited says the write yielded at \
2038 least once before it was done (a prior frame's flush, kernel \
2039 send-buffer pressure or a slow reader)",
2040 );
2041 let timings = &report.timings;
2047 tracing::debug!(
2048 event_name = "session_open",
2049 slot = %slot.display(),
2050 peer = %report.peer,
2051 open_ms = report.open_at.elapsed().as_secs_f64() * 1000.0,
2052 queue_ms = report.queue_ms,
2053 prelude_ms = report.prelude_ms,
2054 queued_ms = report.queued_ms,
2055 hold_ms = report.hold_ms,
2056 prologue_ms = timings.prologue_ms,
2057 header_ms = timings.header_ms,
2058 catch_up_ms = timings.catch_up_ms,
2059 seed_ms = timings.seed_ms,
2060 conversation_ms = timings.conversation_ms,
2061 page_ms = timings.page_ms,
2062 cards_ms = timings.cards_ms,
2063 work_ms = timings.work_ms,
2064 base_ms = timings.base_ms,
2065 rescan_ms = timings.rescan_ms,
2066 content_ms = timings.content_ms,
2067 reviews_ms = timings.reviews_ms,
2068 tree_reviews_ms = timings.tree_reviews_ms,
2069 state_ms = timings.state_ms,
2070 index_ms = timings.index_ms,
2071 resume_ms = timings.resume_ms,
2072 tail_ms = timings.tail_ms,
2073 encode_ms = timings.encode_ms,
2074 pr_age_s = timings.pr_age_s,
2075 "the subscribe's own work, in DISJOINT slices: queue is the wait it \
2076 spent behind an earlier read on this connection, prelude is what the \
2077 subscribe's own handling cost after it, queued the frames already \
2078 waiting for this subject, hold the seat's watch, prologue the \
2079 roster and its asks, header the seat's header, catch_up the \
2080 copy's wait, seed the transcript read a cold seat pays, \
2081 conversation the copy's arrival, page its cut and fold, cards \
2082 the subagent cards, work the tree scan, base the review read's base \
2083 refresh, rescan the scan pinned to the sha it resolved, content the \
2084 diff's changed files, reviews and tree_reviews the review reads, state \
2085 the seat's own state, index the file walk, resume the `/resume` \
2086 list's read, tail the process read \
2087 and the \
2088 record's assembly, and encode the record's value; pr_age_s is \
2089 how old the PR attempt the record carries is, absent until one \
2090 has run; open_ms ends when the snapshot's write returns, so \
2091 frame_write's prep_ms and write_ms are inside it, and what the \
2092 slices and those two do not sum to is the subscribe path's own \
2093 glue",
2094 );
2095}
2096
2097async fn finish_read<S>(
2100 socket: &mut ClientSocket<S>,
2101 state: &Arc<TransportState>,
2102 watching: &mut Watching,
2103 holds: &mut Holds<'_>,
2104 reads: &mut Option<Reading>,
2105 held: &mut Batch,
2106 done: ReadAnswer,
2107) -> anyhow::Result<()>
2108where
2109 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
2110{
2111 let Some(read) = reads.take() else { return Ok(()) };
2112 match done.outcome {
2113 Ok((data, timings)) => {
2114 write_snapshot(
2115 socket,
2116 state,
2117 watching,
2118 done.subject,
2119 data,
2120 Some(OpenReport {
2121 seat: done.seat,
2122 peer: done.peer,
2123 open_at: done.open_at,
2124 queue_ms: done.queue_ms,
2125 prelude_ms: done.prelude_ms,
2126 queued_ms: done.queued_ms,
2127 hold_ms: done.hold_ms,
2128 timings,
2129 }),
2130 )
2131 .await?;
2132 for frame in read.frames {
2134 held.push(Instant::now(), frame);
2135 }
2136 Ok(())
2137 }
2138 Err(refusal) => {
2142 if let Some(slot) = &done.seat {
2146 holds.give_back(slot);
2147 }
2148 send(
2149 socket,
2150 ServerMessage::Error {
2151 what: "subscribe".to_owned(),
2152 why: refusal.to_string(),
2153 seat: done.seat.clone(),
2154 dispatch_id: None,
2155 },
2156 )
2157 .await
2158 }
2159 }
2160}
2161
2162async fn handle_client<S>(
2164 socket: &mut ClientSocket<S>,
2165 state: &Arc<TransportState>,
2166 watching: &mut Watching,
2167 holds: &mut Holds<'_>,
2168 updates: &mut Option<(mpsc::UnboundedReceiver<SeatUpdate>, bool)>,
2169 dictate: &mut Streaming,
2170 viewers: &mut Viewers,
2171 reads: &mut Reads,
2172 arrival: Arrival,
2173 msg: Message,
2174) -> anyhow::Result<()>
2175where
2176 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
2177{
2178 let caller = arrival.caller;
2179 let me = caller.id;
2180 let text = match msg {
2181 Message::Text(text) => text,
2182 Message::Binary(bytes) => {
2186 take_frame(
2187 &state.surface,
2188 dictate,
2189 &mut viewers.terminal,
2190 me,
2191 frame_route(&bytes, dictate),
2192 );
2193 return Ok(());
2194 }
2195 _ => return Ok(()),
2196 };
2197 let Ok(client) = serde_json::from_str::<ClientMessage>(&text) else {
2198 let command = serde_json::from_str::<serde_json::Value>(&text).is_ok_and(|frame| {
2207 frame.get("kind").and_then(serde_json::Value::as_str) == Some("command")
2208 });
2209 let (what, why) = if command {
2210 (
2211 "dispatch",
2212 "this server could not read that command; if this client's build is older than \
2213 the server's, restart the client so both speak the same protocol - the prompt \
2214 it answered is still waiting",
2215 )
2216 } else {
2217 ("client_message", "that is not a message this server knows")
2218 };
2219 return send(
2220 socket,
2221 ServerMessage::Error {
2222 what: what.to_owned(),
2223 why: why.to_owned(),
2224 seat: None,
2225 dispatch_id: None,
2226 },
2227 )
2228 .await;
2229 };
2230 handle_message(
2231 socket, state, watching, holds, updates, dictate, viewers, reads, arrival, client,
2232 )
2233 .await
2234}
2235
2236async fn attach_view<S>(
2247 socket: &mut ClientSocket<S>,
2248 state: &Arc<TransportState>,
2249 viewing: &mut Option<Viewing>,
2250 me: u64,
2251 seat: forge_primitives::SessionSlot,
2252 reply_to: Option<u64>,
2253 tab: Option<u16>,
2254 page: Option<String>,
2255) -> anyhow::Result<()>
2256where
2257 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
2258{
2259 let Some(host) = state.browser_host.as_ref() else {
2260 let why = "this server has no browser to view".to_owned();
2261 return refuse_attach(socket, reply_to, &why, seat).await;
2262 };
2263 let attached = match page.as_deref() {
2266 Some(target) => host.attach_page(me, &seat, target).await,
2267 None => host.attach(me, &seat, tab).await,
2268 };
2269 match attached {
2270 Ok(attachment) => {
2271 let key = attachment.key.clone();
2272 let id = attachment.id;
2273 let tab = attachment.tab;
2274 let mut fresh = Viewing::new(attachment);
2279 let now = *fresh.view.borrow_and_update();
2280 let read_nav = host.nav_state(&key, tab).await;
2281 let watched_nav = *fresh.nav.borrow_and_update();
2284 let nav = read_nav.or(watched_nav);
2285 let kept = fresh.frames.borrow().clone().filter(|frame| frame.tab == tab);
2291 if let Some(frame) = kept.as_ref() {
2292 fresh.last_frame = Some(frame.seq);
2293 }
2294 *viewing = Some(fresh);
2295 send(
2296 socket,
2297 ServerMessage::BrowserView {
2298 id,
2299 profile: key.clone(),
2300 tab,
2301 controller: now.controller,
2302 viewers: now.viewers,
2303 viewer: me,
2304 },
2305 )
2306 .await?;
2307 if let Some(frame) = kept {
2308 let mut transport = super::frame::SocketFrames { socket: &mut *socket };
2309 transport.deliver(&frame).await?;
2310 }
2311 if let Some(nav) = nav {
2312 send(
2313 socket,
2314 ServerMessage::BrowserNavState {
2315 id,
2316 tab,
2317 can_back: nav.can_back,
2318 can_forward: nav.can_forward,
2319 },
2320 )
2321 .await?;
2322 }
2323 match reply_to {
2324 Some(ask) => {
2325 send(
2330 socket,
2331 ServerMessage::Reply {
2332 reply_to: ask,
2333 body: json!({ "ok": true, "profile": key, "tab": tab }),
2334 },
2335 )
2336 .await
2337 }
2338 None => Ok(()),
2339 }
2340 }
2341 Err(why) => {
2342 tracing::debug!(
2348 event_name = "browser_attach_refused",
2349 viewer = me,
2350 requested_tab = tab,
2351 requested_page = page.as_deref(),
2352 was_profile = viewing.as_ref().map(|viewed| viewed.key.as_str()),
2353 was_tab = viewing.as_ref().map(|viewed| viewed.tab),
2354 dropped = viewing.is_some(),
2355 reason = %why,
2356 "an attach was refused",
2357 );
2358 host.detach(me).await;
2362 *viewing = None;
2363 refuse_attach(socket, reply_to, &why, seat).await
2364 }
2365 }
2366}
2367
2368async fn attach_terminal<S>(
2373 socket: &mut ClientSocket<S>,
2374 state: &Arc<TransportState>,
2375 terminal: &mut TerminalViewingState,
2376 me: u64,
2377 seat: forge_primitives::SessionSlot,
2378 id: Option<u16>,
2379 cols: Option<u16>,
2380 rows: Option<u16>,
2381 reply_to: Option<u64>,
2382) -> anyhow::Result<()>
2383where
2384 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
2385{
2386 terminal.viewing = None;
2391 match state.surface.terminal_attach_or_open(&seat, id) {
2392 Ok(entry) => {
2393 if let (Some(cols), Some(rows)) = (cols, rows)
2397 && let Err(why) = entry.terminal.resize(cols, rows)
2398 {
2399 tracing::debug!(
2400 event_name = "terminal_attach_resize_failed",
2401 seat = %seat.display(),
2402 reason = %why,
2403 "an attach's grid was refused",
2404 );
2405 }
2406 let output = entry.terminal.subscribe();
2407 let (repaint, seq) = entry.terminal.repaint_and_seq();
2411 let last_seq = seq.saturating_sub(1);
2412 let id = entry.id;
2415 let control = entry.control_stream();
2416 let viewing = TerminalViewing {
2417 seat: seat.clone(),
2418 id,
2419 entry,
2420 output,
2421 control,
2422 viewer: me,
2423 paused: false,
2424 last_seq,
2425 };
2426 send_terminal_view(socket, &viewing, &repaint).await?;
2427 terminal.viewing = Some(viewing);
2428 match reply_to {
2429 Some(ask) => {
2430 send(
2431 socket,
2432 ServerMessage::Reply {
2433 reply_to: ask,
2434 body: json!({ "ok": true, "id": id }),
2435 },
2436 )
2437 .await
2438 }
2439 None => Ok(()),
2440 }
2441 }
2442 Err(why) => {
2443 tracing::debug!(
2444 event_name = "terminal_attach_refused",
2445 seat = %seat.display(),
2446 id,
2447 reason = %why,
2448 "a terminal attach was refused",
2449 );
2450 match reply_to {
2451 Some(ask) => {
2452 send(
2453 socket,
2454 ServerMessage::Reply {
2455 reply_to: ask,
2456 body: json!({ "ok": false, "why": why }),
2457 },
2458 )
2459 .await
2460 }
2461 None => {
2462 send(
2463 socket,
2464 ServerMessage::Error {
2465 what: "terminal_view".to_owned(),
2466 why,
2467 seat: Some(seat),
2468 dispatch_id: None,
2469 },
2470 )
2471 .await
2472 }
2473 }
2474 }
2475 }
2476}
2477
2478async fn handle_message<S>(
2481 socket: &mut ClientSocket<S>,
2482 state: &Arc<TransportState>,
2483 watching: &mut Watching,
2484 holds: &mut Holds<'_>,
2485 updates: &mut Option<(mpsc::UnboundedReceiver<SeatUpdate>, bool)>,
2486 dictate: &mut Streaming,
2487 viewers: &mut Viewers,
2488 reads: &mut Reads,
2489 arrival: Arrival,
2490 client: ClientMessage,
2491) -> anyhow::Result<()>
2492where
2493 S: futures_util::Sink<Message, Error = axum::Error> + Unpin + Send,
2494{
2495 let caller = arrival.caller;
2496 let arrived_at = arrival.at;
2497 let Viewers { browser, terminal } = viewers;
2498 let viewing = &mut browser.viewing;
2499 let tabs = &mut browser.tabs;
2500 let me = caller.id;
2501 let cap = state.surface.tool_record_cap_bytes();
2504 match client {
2505 ClientMessage::SettingsEdit { fingerprint, edits, id } => {
2506 match state.surface.save_settings(&fingerprint, &edits) {
2507 Ok(()) => send(socket, ServerMessage::SettingsSaved { id }).await?,
2508 Err(error) => {
2509 let (reason, message, key_path) = match error {
2510 forge_workspace::SettingsSaveError::Stale => {
2511 (SettingsRefusal::Stale, error.to_string(), None)
2512 }
2513 forge_workspace::SettingsSaveError::Invalid { message, key_path } => {
2514 (SettingsRefusal::Invalid, message, key_path)
2515 }
2516 forge_workspace::SettingsSaveError::Refused { message, key_path } => {
2517 (SettingsRefusal::Refused, message, key_path)
2518 }
2519 forge_workspace::SettingsSaveError::Io(message) => {
2522 (SettingsRefusal::Io, message, None)
2523 }
2524 };
2525 send(socket, ServerMessage::SettingsRefused { reason, message, id, key_path })
2526 .await?;
2527 }
2528 }
2529 Ok(())
2530 }
2531 ClientMessage::Subscribe { what, answering, review } => {
2532 let open_at = arrived_at;
2539 let wait_ms =
2542 Instant::now().saturating_duration_since(arrived_at).as_secs_f64() * 1000.0;
2543 if let Subject::Session(slot) = &what {
2547 tracing::debug!(
2548 event_name = "client_subscribed",
2549 slot = %slot.display(),
2550 peer = %caller.peer,
2551 "a client subscribed a seat, from the address it is attached on",
2552 );
2553 }
2554 if reads.one.is_some() {
2560 reads.waiting.push_back(Queued::Ask(
2561 arrived_at,
2562 ClientMessage::Subscribe { what, answering, review },
2563 ));
2564 return Ok(());
2565 }
2566 let queued_at = Instant::now();
2569 let queued = forward_queued(
2570 state,
2571 open_stream(state, updates, answering),
2572 &watching.subjects,
2573 caller.id,
2574 );
2575 let mut hold_ms = 0.0;
2583 if let Subject::Session(slot) = &what {
2584 let at = Instant::now();
2590 state.surface.hold_seat(slot);
2591 holds.take(slot);
2592 hold_ms = at.elapsed().as_secs_f64() * 1000.0;
2593 }
2594 batch::flush(socket, queued).await?;
2595 let queued_ms = queued_at.elapsed().as_secs_f64() * 1000.0 - hold_ms;
2599 let prelude_ms =
2603 queued_at.saturating_duration_since(open_at).as_secs_f64() * 1000.0 - wait_ms;
2604 if review {
2613 let seat = match &what {
2614 Subject::Session(slot) => Some(slot.clone()),
2615 _ => None,
2616 };
2617 let state = Arc::clone(state);
2618 let encode_subject = what.clone();
2619 let task = spawn_read(
2620 ReadAnswerGuard {
2621 reply: reads.reply.clone(),
2622 subject: what.clone(),
2623 seat,
2624 peer: caller.peer,
2625 open_at,
2626 queue_ms: wait_ms,
2627 prelude_ms,
2628 queued_ms,
2629 hold_ms,
2630 answered: false,
2631 },
2632 async move { encode_subject_timed(&state, &encode_subject, true).await },
2633 );
2634 reads.one = Some(Reading { subject: what, frames: Vec::new(), _task: task });
2635 return Ok(());
2636 }
2637 match encode_subject_timed(state, &what, review).await {
2638 Ok((data, timings)) => {
2639 let seat = match &what {
2640 Subject::Session(slot) => Some(slot.clone()),
2641 _ => None,
2642 };
2643 write_snapshot(
2644 socket,
2645 state,
2646 watching,
2647 what,
2648 data,
2649 Some(OpenReport {
2650 seat,
2651 peer: caller.peer,
2652 open_at,
2653 queue_ms: wait_ms,
2654 prelude_ms,
2655 queued_ms,
2656 hold_ms,
2657 timings,
2658 }),
2659 )
2660 .await
2661 }
2662 Err(refusal) => {
2666 if let Subject::Session(slot) = &what {
2670 holds.give_back(slot);
2671 }
2672 let seat = match &what {
2677 Subject::Session(slot) => Some(slot.clone()),
2678 _ => None,
2679 };
2680 send(
2681 socket,
2682 ServerMessage::Error {
2683 what: "subscribe".to_owned(),
2684 why: refusal.to_string(),
2685 seat,
2686 dispatch_id: None,
2687 },
2688 )
2689 .await?;
2690 Ok(())
2691 }
2692 }
2693 }
2694 ClientMessage::Command { command, reply_to } => {
2695 let command = stamp(*command, me);
2696 let streamed = match &command {
2702 Command::DictateStream { key, .. } => Some(key.clone()),
2703 _ => None,
2704 };
2705 let recording = match &command {
2706 Command::DictateReadAloudStart { .. } => Some(true),
2707 Command::DictateReadAloudStop { .. } => Some(false),
2708 _ => None,
2709 };
2710 let dispatch_id = command.dispatch_id();
2724 match (reply_to, answers_through_a_reply(&command)) {
2725 (Some(to), true) => {
2726 dispatched(me, &command);
2727 dispatch_answering(state, command, to, &reads.reply);
2728 Ok(())
2729 }
2730 (None, true) => {
2731 send(
2732 socket,
2733 ServerMessage::Error {
2734 what: "reply_to".to_owned(),
2735 why: "this command answers through `reply_to` and no update carries its outcome, so that field is required: without it a client cannot tell a refusal from success".to_owned(),
2736 seat: None,
2737 dispatch_id,
2738 },
2739 )
2740 .await
2741 }
2742 (Some(_), false) => {
2746 send(
2747 socket,
2748 ServerMessage::Error {
2749 what: "reply_to".to_owned(),
2750 why: "this command's outcome rides the subscription rather than a reply, so it is sent without `reply_to`: there is no message a reply for it would carry".to_owned(),
2751 seat: None,
2752 dispatch_id,
2753 },
2754 )
2755 .await
2756 }
2757 (None, false) => {
2758 dispatched(me, &command);
2759 if let Command::ReadCallOutput { key, call_id } = &command {
2765 let output = call_output(state, key, call_id).await;
2766 return send(
2767 socket,
2768 ServerMessage::Update {
2769 update: Box::new(SessionUpdate::CallOutput {
2770 key: key.clone(),
2771 call_id: call_id.clone(),
2772 output,
2773 }),
2774 seq: None,
2775 span: 1,
2776 },
2777 )
2778 .await;
2779 }
2780 match state.surface.dispatch(command) {
2781 Ok(()) => {
2782 if let Some(seat) = streamed {
2791 dictate.claim(&seat);
2792 }
2793 if let Some(recording) = recording {
2794 dictate.read_aloud = recording;
2795 }
2796 Ok(())
2797 }
2798 Err(refusal) => {
2799 send(socket, dispatch_refusal(&refusal, dispatch_id)).await
2800 }
2801 }
2802 }
2803 }
2804 }
2805 ClientMessage::More { conversation, before, turns } => {
2806 let turns = turns.clamp(1, crate::transport::wire::SUBSCRIBE_TURNS);
2812 let roster = state.surface.roster();
2813 let Some(cwd) = roster.cwd_for(&conversation) else {
2817 return send(
2818 socket,
2819 ServerMessage::Error {
2820 what: "more".to_owned(),
2821 why: format!("forge holds no session for {}", conversation.display()),
2822 seat: Some(conversation.clone()),
2826 dispatch_id: None,
2827 },
2828 )
2829 .await;
2830 };
2831 let Some(held) =
2846 conversation_for(state, &conversation, &mut OpenTimings::default()).await
2847 else {
2848 return send(
2856 socket,
2857 ServerMessage::Error {
2858 what: "more".to_owned(),
2859 why: format!(
2860 "the conversation for {} is not held yet, so this page cannot be \
2861 answered; asking again may find it",
2862 conversation.display()
2863 ),
2864 seat: Some(conversation.clone()),
2865 dispatch_id: None,
2866 },
2867 )
2868 .await;
2869 };
2870 let seat = conversation.clone();
2871 let ask = below_the_floor(before.as_deref(), &held);
2879 if matches!(ask, Ask::Below(_) | Ask::BelowWindow) {
2880 let anchors: Vec<Option<forge_primitives::TranscriptAnchor>> = match ask {
2888 Ask::Below(anchor) => vec![Some(anchor)],
2889 Ask::BelowWindow => match boundary_candidates(&held.lock()) {
2890 candidates if candidates.is_empty() => vec![None],
2891 candidates => candidates.into_iter().map(Some).collect(),
2892 },
2893 Ask::Held | Ask::Newest => Vec::new(),
2894 };
2895 for anchor in anchors {
2896 match rows_below_floor(state, &seat, &cwd, anchor, turns).await {
2897 BelowFloor::Page(page) => {
2898 let cron_names = cron_names_off_loop(state, &seat, &page.turns).await;
2899 return send(
2900 socket,
2901 ServerMessage::Page {
2902 conversation,
2903 turns: page.turns,
2904 cursor: page.cursor,
2905 cron_names,
2906 },
2907 )
2908 .await;
2909 }
2910 BelowFloor::Refused => {
2911 return send(
2912 socket,
2913 ServerMessage::Error {
2914 what: "more".to_owned(),
2915 why: format!(
2916 "the fold over {} did not finish, so this page cannot be \
2917 answered; asking again may find it",
2918 seat.display()
2919 ),
2920 seat: Some(seat),
2921 dispatch_id: None,
2922 },
2923 )
2924 .await;
2925 }
2926 BelowFloor::Empty => {
2927 tracing::debug!(
2928 event_name = "page_boundary_unresolved",
2929 slot = %seat.display(),
2930 "a boundary row the transcript does not carry; the walk steps to \
2931 the next stop the window names"
2932 );
2933 }
2934 }
2935 }
2936 return send(
2937 socket,
2938 ServerMessage::Page {
2939 conversation,
2940 turns: Vec::new(),
2941 cursor: None,
2942 cron_names: std::collections::BTreeMap::new(),
2943 },
2944 )
2945 .await;
2946 }
2947 let opening = before.clone();
2948 let floor = held.lock().dropped();
2949 let capped = seat.clone();
2950 let budget = state.surface.more_page_cap_bytes();
2951 let folded = tokio::task::spawn_blocking(move || {
2952 let mut page = held.read(|held| {
2953 page_within(
2954 held.messages(),
2955 held.rendered(),
2956 held.dropped(),
2957 held.dropped_thinking(),
2958 opening.as_deref(),
2959 turns,
2960 budget,
2961 "more_page_cap_kb",
2962 Some((cap, &capped)),
2963 )
2964 });
2965 if budget == usize::MAX {
2971 cap_tool_records(&mut page.turns, cap, &capped);
2972 }
2973 page
2974 })
2975 .await;
2976 let page = match folded {
2982 Ok(mut page) => {
2989 floor_cursor(&mut page, floor);
2990 page
2991 }
2992 Err(err) => {
2993 tracing::warn!(
2994 event_name = "transcript_fold_failed",
2995 slot = %seat.display(),
2996 error = %err,
2997 "the fold did not finish; this page is refused rather than answered empty",
2998 );
2999 return send(
3000 socket,
3001 ServerMessage::Error {
3002 what: "more".to_owned(),
3003 why: format!(
3004 "the fold over {} did not finish, so this page cannot be \
3005 answered; asking again may find it",
3006 seat.display()
3007 ),
3008 seat: Some(seat.clone()),
3009 dispatch_id: None,
3010 },
3011 )
3012 .await;
3013 }
3014 };
3015 let cron_names = cron_names_off_loop(state, &seat, &page.turns).await;
3016 send(
3017 socket,
3018 ServerMessage::Page {
3019 conversation,
3020 turns: page.turns,
3021 cursor: page.cursor,
3022 cron_names,
3023 },
3024 )
3025 .await
3026 }
3027 ClientMessage::SeatRange { slot, epoch, from, to } => {
3030 let Some(cwd) = state.surface.roster().cwd_for(&slot) else {
3031 return send(
3032 socket,
3033 ServerMessage::Error {
3034 what: "seat_range".to_owned(),
3035 why: format!("forge holds no session for {}", slot.display()),
3036 seat: Some(slot),
3037 dispatch_id: None,
3038 },
3039 )
3040 .await;
3041 };
3042 if state.surface.header(&slot).session_id != epoch {
3048 return send(
3049 socket,
3050 ServerMessage::Error {
3051 what: "seat_range".to_owned(),
3052 why: "the seat's occupant has changed since those numbers were minted"
3053 .to_owned(),
3054 seat: Some(slot),
3055 dispatch_id: None,
3056 },
3057 )
3058 .await;
3059 }
3060 let frames = match state.surface.seat_oldest_kept_seq(&slot) {
3061 Some(oldest) if oldest <= from => state
3064 .surface
3065 .seat_frames_between(&slot, from, to)
3066 .into_iter()
3067 .filter_map(|frame| {
3068 Some(SeatFrameWire {
3076 seq: frame.seq?,
3077 update: for_a_client(Arc::unwrap_or_clone(frame.update), cap),
3078 })
3079 })
3080 .collect(),
3081 _ => transcript_frames(state, &slot, &cwd, from, to).await,
3083 };
3084 send(socket, ServerMessage::SeatFrames { slot, epoch, frames }).await
3085 }
3086 ClientMessage::DocumentRead { session, path, reviewable } => {
3087 let surface = Arc::clone(&state.surface);
3092 let seat = session.clone();
3093 let outcome = tokio::task::spawn_blocking(move || {
3094 surface.read_document(&seat, &path, reviewable)
3095 })
3096 .await
3097 .unwrap_or_else(|join| Err(join.to_string()));
3098 send(socket, document_answer(&state.surface, &session, outcome)).await
3099 }
3100 ClientMessage::CommitDiff { session, sha } => {
3101 let Some(cwd) = state.surface.roster().cwd_for(&session) else {
3104 return send(
3105 socket,
3106 ServerMessage::Error {
3107 what: "commit_diff".to_owned(),
3108 why: format!("forge holds no session for {}", session.display()),
3109 seat: Some(session),
3110 dispatch_id: None,
3111 },
3112 )
3113 .await;
3114 };
3115 let content = forge_workspace::env::git_diff::content::scan_commit(&cwd, &sha).await;
3119 send(
3120 socket,
3121 ServerMessage::CommitDiff {
3122 sha,
3123 files: content.files,
3124 truncated: content.truncated,
3125 },
3126 )
3127 .await
3128 }
3129 ClientMessage::FileContent { session, path, sha } => {
3130 let content = match state.surface.file_content(&session, &path, sha.as_deref()).await {
3137 Ok(read) => FileContentWire::Read { lines: read.lines, truncated: read.truncated },
3138 Err(why) => FileContentWire::Failed { why },
3139 };
3140 send(socket, ServerMessage::FileContent { session, path, sha, content }).await
3141 }
3142 ClientMessage::BrowserView { seat, tab, reply_to } => {
3143 attach_view(socket, state, viewing, me, seat, reply_to, tab, None).await
3144 }
3145 ClientMessage::BrowserViewPage { seat, target, reply_to } => {
3146 attach_view(socket, state, viewing, me, seat, reply_to, None, Some(target)).await
3147 }
3148 ClientMessage::TerminalView { seat, id, cols, rows, reply_to } => {
3149 attach_terminal(socket, state, terminal, me, seat, id, cols, rows, reply_to).await
3150 }
3151 ClientMessage::TerminalDetach => {
3152 if let Some((entry, viewer)) = terminal.role.take() {
3158 entry.release_control(viewer);
3159 }
3160 terminal.viewing = None;
3161 Ok(())
3162 }
3163 ClientMessage::TerminalResize { seat, id, cols, rows } => {
3164 match state.surface.terminal_resize(&seat, id, cols, rows) {
3165 Ok(()) => Ok(()),
3166 Err(why) => {
3167 send(
3168 socket,
3169 ServerMessage::Error {
3170 what: "terminal_resize".to_owned(),
3171 why,
3172 seat: Some(seat),
3173 dispatch_id: None,
3174 },
3175 )
3176 .await
3177 }
3178 }
3179 }
3180 ClientMessage::TerminalPause { paused } => {
3181 if let Some(viewing) = terminal.viewing.as_mut() {
3182 viewing.paused = paused;
3183 if !paused {
3184 let (repaint, seq) = viewing.entry.terminal.repaint_and_seq();
3188 viewing.last_seq = seq.saturating_sub(1);
3189 send_terminal_view(socket, viewing, &repaint).await?;
3190 }
3191 }
3192 Ok(())
3193 }
3194 ClientMessage::TerminalControl { seat, id, take } => {
3195 let attached = terminal
3196 .viewing
3197 .as_ref()
3198 .is_some_and(|viewing| viewing.seat == seat && viewing.id == id);
3199 if !attached {
3200 return send(
3204 socket,
3205 ServerMessage::Error {
3206 what: "terminal_control".to_owned(),
3207 why: "this connection is not viewing that terminal".to_owned(),
3208 seat: Some(seat),
3209 dispatch_id: None,
3210 },
3211 )
3212 .await;
3213 }
3214 if take {
3215 if let Err(why) = state.surface.take_terminal_control(&seat, id, me) {
3222 return send(
3223 socket,
3224 ServerMessage::Error {
3225 what: "terminal_control".to_owned(),
3226 why,
3227 seat: Some(seat),
3228 dispatch_id: None,
3229 },
3230 )
3231 .await;
3232 }
3233 let held = terminal.viewing.as_ref().map(|viewing| viewing.entry.clone());
3236 if let Some(entry) = held {
3237 terminal.role = Some((entry, me));
3238 }
3239 } else if let Some(viewing) = terminal.viewing.as_ref()
3240 && viewing.entry.release_control(me)
3241 {
3242 terminal.role = None;
3243 }
3244 Ok(())
3245 }
3246 ClientMessage::TerminalClose { seat, id } => {
3247 match state.surface.terminal_close(&seat, id) {
3248 Ok(()) => {
3249 terminal.viewing = None;
3250 Ok(())
3251 }
3252 Err(why) => {
3253 send(
3254 socket,
3255 ServerMessage::Error {
3256 what: "terminal_close".to_owned(),
3257 why,
3258 seat: Some(seat),
3259 dispatch_id: None,
3260 },
3261 )
3262 .await
3263 }
3264 }
3265 }
3266 ClientMessage::BrowserDetach => {
3267 if let Some(host) = state.browser_host.as_ref() {
3271 host.detach(me).await;
3272 }
3273 *viewing = None;
3274 Ok(())
3275 }
3276 ClientMessage::BrowserInput { seq, event } => {
3277 let (Some(host), Some(viewing)) = (state.browser_host.as_ref(), viewing.as_ref())
3278 else {
3279 let why = "this connection has no browser attached";
3284 if !discrete_gesture(&event) {
3285 return Ok(());
3286 }
3287 tracing::debug!(
3288 event_name = "browser_input_unattached",
3289 viewer = me,
3290 has_browser = state.browser_host.is_some(),
3291 reason = why,
3292 gesture = gesture_kind(&event),
3293 text_chars = text_chars(&event),
3294 "a gesture arrived from a connection with no attached tab",
3295 );
3296 return send(
3297 socket,
3298 ServerMessage::Error {
3299 what: "browser_input".to_owned(),
3300 why: why.to_owned(),
3301 seat: None,
3302 dispatch_id: None,
3303 },
3304 )
3305 .await;
3306 };
3307 match host.input(&viewing.key, viewing.tab, me, seq, &event).await {
3308 Ok(()) => Ok(()),
3309 Err(why) => {
3310 if discrete_gesture(&event) {
3315 tracing::debug!(
3320 event_name = "browser_input_refused",
3321 profile = %viewing.key,
3322 tab = viewing.tab,
3323 viewer = me,
3324 reason = %why,
3325 gesture = gesture_kind(&event),
3326 text_chars = text_chars(&event),
3327 "a gesture was refused before it reached the page",
3328 );
3329 }
3330 send(
3331 socket,
3332 ServerMessage::Error {
3333 what: "browser_input".to_owned(),
3334 why,
3335 seat: None,
3336 dispatch_id: None,
3337 },
3338 )
3339 .await
3340 }
3341 }
3342 }
3343 ClientMessage::BrowserFileAnswer { id, files } => {
3344 let (Some(host), Some(viewing)) = (state.browser_host.as_ref(), viewing.as_ref())
3345 else {
3346 return Ok(());
3347 };
3348 let controlling = viewing.view.borrow().controller == Some(me);
3354 let outstanding = viewing.files.borrow().as_ref().map(|ask| ask.id) == Some(id);
3355 if !controlling || !outstanding {
3356 return send(
3357 socket,
3358 ServerMessage::Error {
3359 what: "browser_file_answer".to_owned(),
3360 why: "this viewer is not the file chooser's to answer".to_owned(),
3361 seat: None,
3362 dispatch_id: None,
3363 },
3364 )
3365 .await;
3366 }
3367 let files = files
3368 .into_iter()
3369 .map(|file| {
3370 let name = file.name;
3371 base64::engine::general_purpose::STANDARD
3372 .decode(file.data)
3373 .map(|bytes| forge_primitives::browser::BrowserUpload { name, bytes })
3374 })
3375 .collect::<Result<Vec<_>, _>>();
3376 let files = match files {
3377 Ok(files) => files,
3378 Err(why) => {
3379 return send(
3380 socket,
3381 ServerMessage::Error {
3382 what: "browser_file_answer".to_owned(),
3383 why: format!("a file answer's bytes are not base64: {why}"),
3384 seat: None,
3385 dispatch_id: None,
3386 },
3387 )
3388 .await;
3389 }
3390 };
3391 match host.answer_files(&viewing.key, viewing.tab, id, files).await {
3392 Ok(()) => Ok(()),
3393 Err(why) => {
3394 send(
3395 socket,
3396 ServerMessage::Error {
3397 what: "browser_file_answer".to_owned(),
3398 why,
3399 seat: None,
3400 dispatch_id: None,
3401 },
3402 )
3403 .await
3404 }
3405 }
3406 }
3407 ClientMessage::BrowserControl { take } => {
3408 let (Some(host), Some(viewing)) = (state.browser_host.as_ref(), viewing.as_ref())
3409 else {
3410 let why = "this connection has no browser attached";
3414 tracing::debug!(
3415 event_name = "browser_control_unattached",
3416 viewer = me,
3417 take,
3418 has_browser = state.browser_host.is_some(),
3419 reason = why,
3420 "a controller change arrived from a connection with no attached tab",
3421 );
3422 return send(
3423 socket,
3424 ServerMessage::Error {
3425 what: "browser_control".to_owned(),
3426 why: why.to_owned(),
3427 seat: None,
3428 dispatch_id: None,
3429 },
3430 )
3431 .await;
3432 };
3433 let (key, tab) = (viewing.key.clone(), viewing.tab);
3434 let outcome = if take {
3435 host.take_control(&key, tab, me).await
3436 } else {
3437 host.release_control(&key, tab, me).await;
3438 Ok(())
3439 };
3440 match outcome {
3441 Ok(()) => {
3442 if take {
3447 tracing::debug!(
3448 event_name = "browser_control",
3449 profile = %key,
3450 tab,
3451 viewer = me,
3452 landed = true,
3453 "the controller role moved",
3454 );
3455 }
3456 Ok(())
3457 }
3458 Err(why) => {
3459 tracing::debug!(
3460 event_name = "browser_control",
3461 profile = %key,
3462 tab,
3463 viewer = me,
3464 take,
3465 landed = false,
3466 reason = %why,
3467 "a controller change was refused",
3468 );
3469 send(
3470 socket,
3471 ServerMessage::Error {
3472 what: "browser_control".to_owned(),
3473 why,
3474 seat: None,
3475 dispatch_id: None,
3476 },
3477 )
3478 .await
3479 }
3480 }
3481 }
3482 ClientMessage::BrowserNav { back } => {
3483 let (Some(host), Some(viewing)) = (state.browser_host.as_ref(), viewing.as_ref())
3484 else {
3485 return Ok(());
3486 };
3487 let (key, id, tab) = (viewing.key.clone(), viewing.id, viewing.tab);
3488 match host.nav(&key, tab, me, back).await {
3489 Ok(nav) => {
3490 send(
3491 socket,
3492 ServerMessage::BrowserNavState {
3493 id,
3494 tab,
3495 can_back: nav.can_back,
3496 can_forward: nav.can_forward,
3497 },
3498 )
3499 .await
3500 }
3501 Err(why) => {
3502 send(
3503 socket,
3504 ServerMessage::Error {
3505 what: "browser_nav".to_owned(),
3506 why,
3507 seat: None,
3508 dispatch_id: None,
3509 },
3510 )
3511 .await
3512 }
3513 }
3514 }
3515 ClientMessage::BrowserTabs { seat, reply_to } => {
3516 let Some(host) = state.browser_host.as_ref() else {
3520 let why = "this server has no browser to list".to_owned();
3521 return answer_ask(socket, reply_to, "browser_tabs", Err(why)).await;
3522 };
3523 match host.tabs(&seat).await {
3524 Ok(list) => {
3525 match host.tab_stream(&seat).await {
3529 Ok(stream) => *tabs = Some(stream),
3530 Err(why) => {
3531 tracing::debug!(
3532 event_name = "browser_tab_stream_failed",
3533 seat = %seat.display(),
3534 %why,
3535 "a tab list was read but its stream could not be followed",
3536 );
3537 }
3538 }
3539 answer_ask(socket, reply_to, "browser_tabs", Ok(json!({ "tabs": list }))).await
3540 }
3541 Err(why) => answer_ask(socket, reply_to, "browser_tabs", Err(why)).await,
3542 }
3543 }
3544 ClientMessage::BrowserResize { width, height, scale } => {
3545 let (Some(host), Some(viewing)) = (state.browser_host.as_ref(), viewing.as_ref())
3550 else {
3551 return Ok(());
3552 };
3553 let (key, tab) = (viewing.key.clone(), viewing.tab);
3554 host.resize(&key, tab, me, width, height, scale).await;
3555 Ok(())
3556 }
3557 ClientMessage::ClientPerf { records } => {
3558 let count = records.len();
3564 match tokio::task::spawn_blocking(move || crate::perf_log::append(&records)).await {
3565 Ok(Ok(())) => {}
3566 Ok(Err(why)) => tracing::debug!(
3567 event_name = "client_perf_write_failed",
3568 records = count,
3569 %why,
3570 "a client's perf batch could not be written",
3571 ),
3572 Err(why) => tracing::debug!(
3573 event_name = "client_perf_write_failed",
3574 records = count,
3575 %why,
3576 "a client's perf batch could not be written",
3577 ),
3578 }
3579 Ok(())
3580 }
3581 ClientMessage::Unsubscribe { what } => {
3582 if let Some(at) = watching.subjects.iter().position(|held| held == &what) {
3594 watching.subjects.remove(at);
3595 if let Subject::Session(slot) = &what {
3596 holds.give_back(slot);
3600 }
3601 }
3602 if reads.one.as_ref().is_some_and(|read| read.subject == what) {
3611 let _ = reads.one.take();
3612 if let Subject::Session(slot) = &what {
3613 holds.give_back(slot);
3614 }
3615 send(socket, released_subscribe(&what)).await?;
3616 }
3617 let mut kept = std::collections::VecDeque::with_capacity(reads.waiting.len());
3625 while let Some(entry) = reads.waiting.pop_front() {
3626 match entry {
3627 Queued::Ask(_, ClientMessage::Subscribe { what: queued, .. })
3628 if queued == what =>
3629 {
3630 kept.push_back(Queued::Withdrawn(what.clone()));
3631 }
3632 other => kept.push_back(other),
3633 }
3634 }
3635 reads.waiting = kept;
3636 Ok(())
3637 }
3638 ClientMessage::Showing { seat } => set_showing(socket, state, watching, seat).await,
3639 ClientMessage::Ping => send(socket, ServerMessage::Pong).await,
3644 }
3645}
3646
3647fn released_subscribe(what: &Subject) -> ServerMessage {
3650 ServerMessage::Error {
3651 what: "subscribe".to_owned(),
3652 why: "the seat was released before its read answered".to_owned(),
3653 seat: match what {
3654 Subject::Session(slot) => Some(slot.clone()),
3655 _ => None,
3656 },
3657 dispatch_id: None,
3658 }
3659}
3660
3661async fn set_showing<S>(
3669 socket: &mut ClientSocket<S>,
3670 state: &Arc<TransportState>,
3671 watching: &mut Watching,
3672 seat: Option<SessionSlot>,
3673) -> anyhow::Result<()>
3674where
3675 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
3676{
3677 if watching.showing == seat {
3678 return Ok(());
3679 }
3680 if watching.showing_attached {
3681 if let Some(previous) = watching.showing.as_ref() {
3682 Live::lock(&state.live).detach(previous);
3683 }
3684 watching.showing_attached = false;
3685 }
3686 let Some(next) = seat else {
3687 watching.showing = None;
3688 socket.note_showing(None);
3689 return Ok(());
3690 };
3691 let watched = watching
3697 .subjects
3698 .iter()
3699 .any(|held| matches!(held, Subject::Session(slot) if slot == &next));
3700 watching.showing = Some(next.clone());
3701 socket.note_showing(Some(&next));
3702 if watched {
3703 spend_showing(socket, state, watching, &next).await?;
3704 }
3705 Ok(())
3706}
3707
3708async fn spend_showing<S>(
3713 socket: &mut ClientSocket<S>,
3714 state: &Arc<TransportState>,
3715 watching: &mut Watching,
3716 slot: &SessionSlot,
3717) -> anyhow::Result<()>
3718where
3719 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
3720{
3721 let failed = state.surface.failed_turn(slot);
3722 let death = state.surface.seat_shown(slot);
3725 let spent = Live::lock(&state.live).attach(slot, failed) || death;
3726 watching.showing_attached = true;
3727 if spent && watching.subjects.iter().any(|held| matches!(held, Subject::Home)) {
3728 match encode_subject(state, &Subject::Home).await {
3729 Ok(home) => {
3730 send(socket, ServerMessage::Snapshot { subject: Subject::Home, data: home })
3731 .await?;
3732 }
3733 Err(error) => tracing::debug!(
3734 event_name = "home_refresh_failed",
3735 slot = %slot.display(),
3736 %error,
3737 "the home refresh after a seat was shown could not be encoded",
3738 ),
3739 }
3740 }
3741 Ok(())
3742}
3743
3744fn mint_connection_id() -> u64 {
3754 static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
3755 NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
3756}
3757
3758fn discrete_gesture(event: &BrowserInputEvent) -> bool {
3762 !matches!(
3763 event,
3764 BrowserInputEvent::Mouse { action: MouseAction::Move, .. }
3765 | BrowserInputEvent::Wheel { .. }
3766 )
3767}
3768
3769fn gesture_kind(event: &BrowserInputEvent) -> &'static str {
3772 match event {
3773 BrowserInputEvent::Mouse { action: MouseAction::Press, .. } => "mouse press",
3774 BrowserInputEvent::Mouse { action: MouseAction::Release, .. } => "mouse release",
3775 BrowserInputEvent::Mouse { action: MouseAction::Move, .. } => "mouse move",
3776 BrowserInputEvent::Wheel { .. } => "wheel",
3777 BrowserInputEvent::Key { down: true, .. } => "key down",
3778 BrowserInputEvent::Key { down: false, .. } => "key up",
3779 BrowserInputEvent::Text { .. } => "text",
3780 }
3781}
3782
3783fn text_chars(event: &BrowserInputEvent) -> Option<usize> {
3785 match event {
3786 BrowserInputEvent::Text { text } => Some(text.chars().count()),
3787 _ => None,
3788 }
3789}
3790
3791async fn transcript_frames(
3802 state: &TransportState,
3803 slot: &SessionSlot,
3804 cwd: &Path,
3805 from: u64,
3806 to: u64,
3807) -> Vec<SeatFrameWire> {
3808 if to < from {
3809 return Vec::new();
3810 }
3811 let want = (to - from).saturating_add(1);
3812 let surface = Arc::clone(&state.surface);
3813 let seat = slot.clone();
3814 let cwd = cwd.to_owned();
3815 let Ok(read) = tokio::task::spawn_blocking(move || surface.conversation(&seat, &cwd)).await
3819 else {
3820 return Vec::new();
3821 };
3822 let rows = &read.messages;
3823 let kept = u64::try_from(rows.len()).unwrap_or(u64::MAX).min(want);
3824 let first = to.saturating_add(1).saturating_sub(kept);
3825 let cap = state.surface.tool_record_cap_bytes();
3828 rows.iter()
3829 .skip(rows.len().saturating_sub(usize::try_from(kept).unwrap_or(usize::MAX)))
3830 .enumerate()
3831 .map(|(at, msg)| {
3832 let mut msg = msg.clone();
3833 cap_message_record(&mut msg, cap, slot);
3834 SeatFrameWire {
3835 seq: first.saturating_add(u64::try_from(at).unwrap_or(u64::MAX)),
3836 update: SessionUpdate::ChatAppended { key: slot.clone(), source: None, msg },
3837 }
3838 })
3839 .collect()
3840}
3841
3842fn ours_to_hear(update: &SessionUpdate, me: u64) -> bool {
3850 match update {
3851 SessionUpdate::DictateStarted { initiator, .. }
3852 | SessionUpdate::DictateLevel { initiator, .. }
3853 | SessionUpdate::DictateTranscribing { initiator, .. }
3854 | SessionUpdate::DictateProgress { initiator, .. }
3855 | SessionUpdate::DictateEnded { initiator, .. } => *initiator == Some(me),
3856 _ => true,
3857 }
3858}
3859
3860fn is_take_news(update: &SessionUpdate) -> bool {
3865 matches!(
3866 update,
3867 SessionUpdate::DictateStarted { .. }
3868 | SessionUpdate::DictateLevel { .. }
3869 | SessionUpdate::DictateTranscribing { .. }
3870 | SessionUpdate::DictateProgress { .. }
3871 | SessionUpdate::DictateEnded { .. }
3872 )
3873}
3874
3875fn forwards(update: &SessionUpdate, watched: &[Subject], me: u64) -> bool {
3884 ours_to_hear(update, me)
3885 && (is_take_news(update) || watched.iter().any(|what| what.covers(update)))
3886}
3887
3888fn for_a_client(mut update: SessionUpdate, cap: usize) -> SessionUpdate {
3911 match &mut update {
3912 SessionUpdate::Connected { history, .. }
3913 | SessionUpdate::SessionReplaced { history, .. } => history.clear(),
3914 SessionUpdate::ChatAppended { msg, key, .. } => cap_message_record(msg, cap, key),
3918 _ => {}
3919 }
3920 update
3921}
3922
3923fn forward_queued(
3926 state: &TransportState,
3927 frames: Vec<SeatUpdate>,
3928 watched: &[Subject],
3929 me: u64,
3930) -> Vec<SeatUpdate> {
3931 let cap = state.surface.tool_record_cap_bytes();
3932 frames
3933 .into_iter()
3934 .filter(|frame| forwards(&frame.update, watched, me))
3935 .map(|frame| SeatUpdate {
3936 seq: frame.seq,
3937 update: Arc::new(for_a_client(Arc::unwrap_or_clone(frame.update), cap)),
3938 })
3939 .collect()
3940}
3941
3942#[derive(Debug, PartialEq)]
3944enum FrameRoute {
3945 Audio(Vec<f32>),
3948 TerminalInput { id: u16, bytes: Vec<u8> },
3950 Refused(super::frame::Refusal),
3952 NoTake,
3955}
3956
3957fn frame_route(bytes: &[u8], dictate: &Streaming) -> FrameRoute {
3967 let decoded = match super::frame::decode(bytes) {
3968 Ok(decoded) => decoded,
3969 Err(refusal) => return FrameRoute::Refused(refusal),
3970 };
3971 match decoded {
3972 super::frame::Frame::Audio(samples) => {
3973 if !dictate.any() {
3974 return FrameRoute::NoTake;
3975 }
3976 FrameRoute::Audio(samples)
3977 }
3978 super::frame::Frame::TerminalInput { id, bytes } => FrameRoute::TerminalInput { id, bytes },
3979 }
3980}
3981
3982fn take_frame(
3990 surface: &ViewSurface,
3991 dictate: &Streaming,
3992 terminal: &mut TerminalViewingState,
3993 me: u64,
3994 route: FrameRoute,
3995) {
3996 match route {
3997 FrameRoute::TerminalInput { id, bytes } => {
3998 let Some(viewing) = terminal.viewing.as_ref() else {
4004 tracing::debug!(
4005 event_name = "terminal_input_unattached",
4006 id,
4007 "a terminal input frame arrived on a connection viewing no terminal",
4008 );
4009 return;
4010 };
4011 if viewing.id != id {
4012 tracing::debug!(
4013 event_name = "terminal_input_wrong_terminal",
4014 id,
4015 attached = viewing.id,
4016 "a terminal input frame named a terminal this connection is not viewing",
4017 );
4018 return;
4019 }
4020 if viewing.entry.controller() == Some(me) {
4026 if let Err(why) = surface.terminal_write(&viewing.seat, id, &bytes) {
4027 tracing::debug!(
4028 event_name = "terminal_input_failed",
4029 id,
4030 reason = %why,
4031 "a terminal input frame could not be written",
4032 );
4033 }
4034 return;
4035 }
4036 tracing::debug!(
4037 event_name = "terminal_input_not_controlling",
4038 id,
4039 "a terminal input frame arrived from a connection that does not hold the keyboard",
4040 );
4041 }
4042 FrameRoute::Audio(samples) => {
4043 let kept = dictate
4047 .seats
4048 .iter()
4049 .any(|claim| surface.dictate_push(&claim.seat, &samples, Some(me)))
4050 || (dictate.read_aloud && surface.dictate_read_aloud_push(&samples, Some(me)));
4051 if !kept {
4052 tracing::debug!(
4053 event_name = "dictate_frame_dropped",
4054 "a dictation frame arrived for takes that have stopped or are another connection's",
4055 );
4056 }
4057 }
4058 FrameRoute::Refused(refusal) => tracing::debug!(
4059 event_name = "binary_frame_refused",
4060 reason = %refusal.reason(),
4061 "a binary message was not a frame this server takes",
4062 ),
4063 FrameRoute::NoTake => tracing::debug!(
4064 event_name = "dictate_frame_without_a_take",
4065 "a dictation frame arrived on a connection that has not started a take",
4066 ),
4067 }
4068}
4069
4070const COMPRESSED_MIN_BYTES: usize = 64 * 1024;
4078
4079struct ClientSocket<S> {
4088 inner: S,
4089 deflate: bool,
4090 progress: Option<Arc<Progress>>,
4095}
4096
4097impl<S> ClientSocket<S> {
4098 fn new(inner: S, deflate: bool) -> Self {
4099 Self { inner, deflate, progress: None }
4100 }
4101
4102 fn watched(mut self, progress: Arc<Progress>) -> Self {
4104 self.progress = Some(progress);
4105 self
4106 }
4107
4108 fn note_showing(&self, seat: Option<&SessionSlot>) {
4111 if let Some(progress) = self.progress.as_deref() {
4112 progress.note_showing(seat);
4113 }
4114 }
4115}
4116
4117impl<S> futures_util::Stream for ClientSocket<S>
4118where
4119 S: futures_util::Stream<Item = Result<Message, axum::Error>> + Unpin,
4120{
4121 type Item = Result<Message, axum::Error>;
4122
4123 fn poll_next(
4124 mut self: std::pin::Pin<&mut Self>,
4125 cx: &mut std::task::Context<'_>,
4126 ) -> std::task::Poll<Option<Self::Item>> {
4127 std::pin::Pin::new(&mut self.inner).poll_next(cx)
4128 }
4129}
4130
4131impl<S> futures_util::Sink<Message> for ClientSocket<S>
4132where
4133 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
4134{
4135 type Error = axum::Error;
4136
4137 fn poll_ready(
4138 mut self: std::pin::Pin<&mut Self>,
4139 cx: &mut std::task::Context<'_>,
4140 ) -> std::task::Poll<Result<(), axum::Error>> {
4141 std::pin::Pin::new(&mut self.inner).poll_ready(cx)
4142 }
4143
4144 fn start_send(mut self: std::pin::Pin<&mut Self>, item: Message) -> Result<(), axum::Error> {
4145 let deflate = self.deflate;
4152 let item = match item {
4153 Message::Text(text) if deflate && text.len() >= COMPRESSED_MIN_BYTES => {
4159 frame_for_flag(deflate, text.as_str())
4160 }
4161 other => other,
4162 };
4163 std::pin::Pin::new(&mut self.inner).start_send(item)
4164 }
4165
4166 fn poll_flush(
4167 mut self: std::pin::Pin<&mut Self>,
4168 cx: &mut std::task::Context<'_>,
4169 ) -> std::task::Poll<Result<(), axum::Error>> {
4170 let flushed = std::pin::Pin::new(&mut self.inner).poll_flush(cx);
4171 if matches!(flushed, std::task::Poll::Ready(Ok(())))
4175 && let Some(progress) = self.progress.as_deref()
4176 {
4177 progress.stamp();
4178 }
4179 flushed
4180 }
4181
4182 fn poll_close(
4183 mut self: std::pin::Pin<&mut Self>,
4184 cx: &mut std::task::Context<'_>,
4185 ) -> std::task::Poll<Result<(), axum::Error>> {
4186 std::pin::Pin::new(&mut self.inner).poll_close(cx)
4187 }
4188}
4189
4190fn frame_bytes(frame: &Message) -> usize {
4192 match frame {
4193 Message::Text(text) => text.len(),
4194 Message::Binary(bytes) => bytes.len(),
4195 _ => 0,
4196 }
4197}
4198
4199fn frame_for<S>(socket: &ClientSocket<S>, text: &str) -> Message {
4203 frame_for_flag(socket.deflate, text)
4204}
4205
4206fn frame_for_flag(deflate: bool, text: &str) -> Message {
4207 if !deflate || text.len() < COMPRESSED_MIN_BYTES {
4208 return Message::Text(text.to_owned().into());
4209 }
4210 match gzipped(text.as_bytes()) {
4211 Ok(body) => {
4212 let mut frame = Vec::with_capacity(body.len() + 1);
4213 frame.push(COMPRESSED_JSON_TAG);
4214 frame.extend_from_slice(&body);
4215 Message::Binary(frame.into())
4216 }
4217 Err(_) => Message::Text(text.to_owned().into()),
4220 }
4221}
4222
4223fn gzipped(bytes: &[u8]) -> std::io::Result<Vec<u8>> {
4224 use std::io::Write as _;
4225 let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
4226 encoder.write_all(bytes)?;
4227 encoder.finish()
4228}
4229
4230async fn send<S>(socket: &mut ClientSocket<S>, message: ServerMessage) -> anyhow::Result<()>
4231where
4232 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
4233{
4234 send_frame(socket, message).await.map(|_| ())
4235}
4236
4237async fn send_text<S>(socket: &mut ClientSocket<S>, text: &str) -> anyhow::Result<()>
4241where
4242 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
4243{
4244 let frame = frame_for(socket, text);
4245 write_within(WRITE_BOUND, async { socket.send(frame).await.map_err(anyhow::Error::from) }).await
4246}
4247
4248async fn answer_ask<S>(
4251 socket: &mut ClientSocket<S>,
4252 reply_to: Option<u64>,
4253 what: &str,
4254 outcome: Result<serde_json::Value, String>,
4255) -> anyhow::Result<()>
4256where
4257 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
4258{
4259 match (reply_to, outcome) {
4260 (Some(id), Ok(mut body)) => {
4261 if let Some(map) = body.as_object_mut() {
4262 map.insert("ok".to_owned(), json!(true));
4263 }
4264 send(socket, ServerMessage::Reply { reply_to: id, body }).await
4265 }
4266 (Some(id), Err(why)) => {
4267 send(
4268 socket,
4269 ServerMessage::Reply { reply_to: id, body: json!({ "ok": false, "why": why }) },
4270 )
4271 .await
4272 }
4273 (None, Ok(_)) => Ok(()),
4274 (None, Err(why)) => {
4275 send(
4276 socket,
4277 ServerMessage::Error { what: what.to_owned(), why, seat: None, dispatch_id: None },
4278 )
4279 .await
4280 }
4281 }
4282}
4283
4284async fn refuse_attach<S>(
4287 socket: &mut ClientSocket<S>,
4288 reply_to: Option<u64>,
4289 why: &str,
4290 seat: SessionSlot,
4291) -> anyhow::Result<()>
4292where
4293 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
4294{
4295 match reply_to {
4296 Some(reply_to) => {
4297 send(
4298 socket,
4299 ServerMessage::Reply { reply_to, body: json!({ "ok": false, "why": why }) },
4300 )
4301 .await
4302 }
4303 None => {
4304 send(
4305 socket,
4306 ServerMessage::Error {
4307 what: "browser_view".to_owned(),
4308 why: why.to_owned(),
4309 seat: Some(seat),
4310 dispatch_id: None,
4311 },
4312 )
4313 .await
4314 }
4315 }
4316}
4317
4318async fn send_frame<S>(
4324 socket: &mut ClientSocket<S>,
4325 message: ServerMessage,
4326) -> anyhow::Result<usize>
4327where
4328 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
4329{
4330 let text = serde_json::to_string(&message)?;
4331 let frame = frame_for(socket, &text);
4332 let bytes = frame_bytes(&frame);
4333 write_within(WRITE_BOUND, async { socket.send(frame).await.map_err(anyhow::Error::from) })
4334 .await?;
4335 Ok(bytes)
4336}
4337
4338struct FrameWrite {
4341 bytes: usize,
4342 prep_ms: f64,
4345 write_ms: f64,
4346 waited: bool,
4347}
4348
4349trait WriteClock {
4355 fn start(&self, frame: &Message) -> Instant;
4356
4357 fn end(&self) -> Instant;
4358}
4359
4360struct SystemClock;
4362
4363impl WriteClock for SystemClock {
4364 fn start(&self, _frame: &Message) -> Instant {
4365 Instant::now()
4366 }
4367
4368 fn end(&self) -> Instant {
4369 Instant::now()
4370 }
4371}
4372
4373async fn send_frame_timed<S, C>(
4380 socket: &mut ClientSocket<S>,
4381 message: ServerMessage,
4382 clock: &C,
4383) -> anyhow::Result<FrameWrite>
4384where
4385 S: futures_util::Sink<Message, Error = axum::Error> + Unpin,
4386 C: WriteClock,
4387{
4388 let prep_at = Instant::now();
4389 let text = serde_json::to_string(&message)?;
4390 let frame = frame_for(socket, &text);
4394 let prep_ms = prep_at.elapsed().as_secs_f64() * 1000.0;
4395 let bytes = frame_bytes(&frame);
4396 let waited = AtomicBool::new(false);
4397 let at = clock.start(&frame);
4398 let mut write = std::pin::pin!(async { socket.send(frame).await.map_err(anyhow::Error::from) });
4399 write_within(WRITE_BOUND, async {
4400 std::future::poll_fn(|cx| {
4401 let poll = std::future::Future::poll(write.as_mut(), cx);
4402 if poll.is_pending() {
4403 waited.store(true, Ordering::Relaxed);
4404 }
4405 poll
4406 })
4407 .await
4408 })
4409 .await?;
4410 Ok(FrameWrite {
4411 bytes,
4412 prep_ms,
4413 write_ms: clock.end().saturating_duration_since(at).as_secs_f64() * 1000.0,
4414 waited: waited.load(Ordering::Relaxed),
4415 })
4416}
4417
4418#[derive(Debug)]
4420enum Ask {
4421 Held,
4423 Newest,
4425 Below(forge_primitives::TranscriptAnchor),
4427 BelowWindow,
4430}
4431
4432fn below_the_floor(before: Option<&str>, held: &crate::transport::conversation::Held) -> Ask {
4441 let Some(cursor) = before else {
4442 return Ask::Newest;
4443 };
4444 if let Some(anchor) = row_cursor(cursor) {
4445 return Ask::Below(anchor);
4446 }
4447 match cursor.parse::<usize>() {
4448 Ok(named) if named <= held.lock().dropped() => Ask::BelowWindow,
4449 _ => Ask::Held,
4450 }
4451}
4452
4453fn row_cursor(cursor: &str) -> Option<forge_primitives::TranscriptAnchor> {
4457 let rest = cursor.strip_prefix("r:")?;
4458 let (offset, row) = rest.split_once(':').unwrap_or((rest, ""));
4459 let offset = offset.parse::<u64>().ok()?;
4460 Some(forge_primitives::TranscriptAnchor { row: row.to_owned(), offset: Some(offset) })
4461}
4462
4463fn row_cursor_of(offset: u64, row: &str) -> String {
4465 if row.is_empty() { format!("r:{offset}") } else { format!("r:{offset}:{row}") }
4466}
4467
4468const BOUNDARY_TRIES: usize = 8;
4476
4477async fn cron_names_off_loop(
4483 state: &TransportState,
4484 seat: &SessionSlot,
4485 turns: &[crate::transport::wire::TurnWire],
4486) -> std::collections::BTreeMap<String, forge_primitives::cron::CronName> {
4487 let surface = Arc::clone(&state.surface);
4488 let named = seat.clone();
4489 let seat = seat.clone();
4490 let turns = turns.to_vec();
4491 tokio::task::spawn_blocking(move || {
4492 crate::transport::wire::cron_names_of(&surface, &seat, &turns)
4493 })
4494 .await
4495 .unwrap_or_else(|error| {
4496 tracing::warn!(
4497 event_name = "cron_names_lookup_failed",
4498 slot = %named.display(),
4499 %error,
4500 "the page's cron pairings could not be read; its rows fall back to the prompt"
4501 );
4502 std::collections::BTreeMap::new()
4503 })
4504}
4505
4506fn boundary_candidates(
4525 held: &crate::transport::conversation::Conversation,
4526) -> Vec<forge_primitives::TranscriptAnchor> {
4527 let rows: Vec<&forge_primitives::Message> = held
4528 .messages()
4529 .iter()
4530 .filter(|message| held.has_a_row(message) && row_id(message).is_some())
4531 .collect();
4532 let stops: Vec<&forge_primitives::Message> =
4533 match rows.iter().position(|m| !carries_a_result(m)) {
4534 Some(at) => rows[at..].iter().copied().filter(|m| !carries_a_result(m)).collect(),
4535 None => rows.last().copied().into_iter().collect(),
4536 };
4537 stops
4538 .into_iter()
4539 .filter_map(|message| {
4540 row_id(message).map(|row| forge_primitives::TranscriptAnchor { row, offset: None })
4541 })
4542 .take(BOUNDARY_TRIES)
4543 .collect()
4544}
4545
4546fn carries_a_result(message: &forge_primitives::Message) -> bool {
4549 let forge_primitives::Message::User { message, .. } = message else {
4550 return false;
4551 };
4552 message
4553 .content
4554 .iter()
4555 .any(|block| matches!(block, forge_primitives::ContentBlock::ToolResult { .. }))
4556}
4557
4558fn row_id(message: &forge_primitives::Message) -> Option<String> {
4560 match message {
4561 forge_primitives::Message::User { uuid, .. }
4562 | forge_primitives::Message::Assistant { uuid, .. } => uuid.clone(),
4563 forge_primitives::Message::StopHookSummary { uuid, .. }
4564 | forge_primitives::Message::CompactBoundary { uuid, .. } => Some(uuid.clone()),
4565 _ => None,
4566 }
4567}
4568
4569enum BelowFloor {
4571 Page(crate::transport::wire::Page),
4573 Empty,
4577 Refused,
4581}
4582
4583async fn rows_below_floor(
4598 state: &TransportState,
4599 seat: &SessionSlot,
4600 cwd: &std::path::Path,
4601 below: Option<forge_primitives::TranscriptAnchor>,
4602 turns: u32,
4603) -> BelowFloor {
4604 let surface = Arc::clone(&state.surface);
4605 let session = seat.clone();
4606 let cwd = cwd.to_path_buf();
4607 let rows = (turns as usize)
4610 .saturating_mul(forge_workspace::userdata::catalog::scan::SPAN_ROWS_PER_TURN);
4611 let read = tokio::task::spawn_blocking(move || {
4612 surface.transcript_rows(&session, &cwd, below.as_ref(), rows)
4613 })
4614 .await;
4615 let Some(span) = read.ok().flatten() else {
4616 tracing::debug!(
4617 event_name = "transcript_page_unavailable",
4618 slot = %seat.display(),
4619 "the transcript has no rows for this anchor; the walk decides what the empty read \
4620 means",
4621 );
4622 return BelowFloor::Empty;
4623 };
4624 let live = state.conversations.live_of(seat);
4629 let cap = state.surface.tool_record_cap_bytes();
4630 let budget = state.surface.more_page_cap_bytes();
4631 let capped = seat.clone();
4632 let folded = tokio::task::spawn_blocking(move || {
4633 let mut page = page_of_span(&span, turns, live, budget, Some((cap, &capped)));
4634 if budget == usize::MAX {
4637 cap_tool_records(&mut page.turns, cap, &capped);
4638 }
4639 page
4640 })
4641 .await;
4642 if let Ok(page) = folded {
4643 BelowFloor::Page(page)
4644 } else {
4645 tracing::warn!(
4646 event_name = "transcript_page_fold_failed",
4647 slot = %seat.display(),
4648 "the fold over the transcript span did not finish; this page is refused rather than \
4649 answered empty",
4650 );
4651 BelowFloor::Refused
4652 }
4653}
4654
4655fn page_of_span(
4664 span: &forge_primitives::TranscriptSpan,
4665 turns: u32,
4666 live: bool,
4667 budget: usize,
4668 cap: Option<(usize, &SessionSlot)>,
4669) -> crate::transport::wire::Page {
4670 let rendered = crate::transcript::render(&span.messages, live);
4671 let mut page =
4672 page_within(&span.messages, &rendered, 0, 0, None, turns, budget, "more_page_cap_kb", cap);
4673 let anchor = page.turns.iter().flat_map(|turn| turn.messages.iter()).find_map(|frame| {
4678 let uuid = frame["uuid"].as_str().filter(|uuid| !uuid.is_empty())?;
4679 let at =
4680 span.messages.iter().position(|message| row_id(message).as_deref() == Some(uuid))?;
4681 Some((at, uuid))
4682 });
4683 page.cursor = match anchor {
4684 Some((at, uuid)) if !(at == 0 && span.exhausted) => {
4685 Some(row_cursor_of(span.offsets[at], uuid))
4686 }
4687 _ => None,
4688 };
4689 page
4690}
4691
4692fn dispatch_answering(
4700 state: &TransportState,
4701 command: Command,
4702 to: u64,
4703 reply_tx: &mpsc::UnboundedSender<Answer>,
4704) {
4705 match command {
4719 Command::SpawnWorker {
4720 project_key,
4721 label,
4722 charter,
4723 spawned_by,
4724 resume_existing,
4725 kick,
4726 resume_kick,
4727 interactive,
4728 mcp_families,
4729 cwd,
4730 from_boot_respawn,
4731 ..
4732 } => {
4733 let (tx, rx) = oneshot::channel();
4734 let command = Command::SpawnWorker {
4735 project_key,
4736 label,
4737 charter,
4738 spawned_by,
4739 resume_existing,
4740 kick,
4741 resume_kick,
4742 interactive,
4743 mcp_families,
4744 cwd,
4745 from_boot_respawn,
4746 return_to: Some(tx),
4747 };
4748 answered(state, command, to, rx, reply_tx);
4749 }
4750 Command::DespawnWorker { project_key, label, force, .. } => {
4751 let (tx, rx) = oneshot::channel();
4752 let command = Command::DespawnWorker { project_key, label, force, respond: Some(tx) };
4753 answered(state, command, to, rx, reply_tx);
4754 }
4755 Command::UpsertReviewThread { project, branch, thread, origin, sent_words, .. } => {
4756 let (tx, rx) = oneshot::channel();
4757 let command = Command::UpsertReviewThread {
4758 project,
4759 branch,
4760 thread,
4761 origin,
4762 sent_words,
4763 respond: Some(tx),
4764 };
4765 answered(state, command, to, rx, reply_tx);
4766 }
4767 Command::DeleteReviewThread { project, branch, thread_id, .. } => {
4768 let (tx, rx) = oneshot::channel();
4769 let command =
4770 Command::DeleteReviewThread { project, branch, thread_id, respond: Some(tx) };
4771 answered(state, command, to, rx, reply_tx);
4772 }
4773 Command::SubmitReview { project, branch, summary, thread_ids, origin, .. } => {
4774 let (tx, rx) = oneshot::channel();
4775 let command = Command::SubmitReview {
4776 project,
4777 branch,
4778 summary,
4779 thread_ids,
4780 origin,
4781 respond: Some(tx),
4782 };
4783 answered(state, command, to, rx, reply_tx);
4784 }
4785 Command::BiometricRelease { enrolment, purpose, .. } => {
4786 let (tx, rx) = oneshot::channel();
4787 let command = Command::BiometricRelease { enrolment, purpose, respond: Some(tx) };
4788 answered(state, command, to, rx, reply_tx);
4789 }
4790 Command::BiometricEnrol { enrolment, device, purpose, blob, .. } => {
4791 let (tx, rx) = oneshot::channel();
4792 let command =
4793 Command::BiometricEnrol { enrolment, device, purpose, blob, respond: Some(tx) };
4794 answered(state, command, to, rx, reply_tx);
4795 }
4796 Command::BiometricRewrite { enrolment, purpose, blob, .. } => {
4797 let (tx, rx) = oneshot::channel();
4798 let command = Command::BiometricRewrite { enrolment, purpose, blob, respond: Some(tx) };
4799 answered(state, command, to, rx, reply_tx);
4800 }
4801 Command::VaultEnrolmentMaterial { .. } => {
4802 let (tx, rx) = oneshot::channel();
4803 let command = Command::VaultEnrolmentMaterial { respond: Some(tx) };
4804 answered(state, command, to, rx, reply_tx);
4805 }
4806 other => {
4810 debug_assert!(
4815 !answers_through_a_reply(&other),
4816 "a reply-carrying command reached the fire-and-forget arm",
4817 );
4818 if let Err(refusal) = state.surface.dispatch(other) {
4819 let body = serde_json::Value::String(refusal.to_string());
4820 let _ = reply_tx
4821 .send(Answer::Reply(Box::new(ServerMessage::Reply { reply_to: to, body })));
4822 }
4823 }
4824 }
4825}
4826
4827fn document_answer(
4832 surface: &ViewSurface,
4833 seat: &SessionSlot,
4834 outcome: Result<DocumentRead, String>,
4835) -> ServerMessage {
4836 let read = match outcome {
4837 Ok(read) => read,
4838 Err(why) => {
4839 return ServerMessage::Error {
4840 what: "document_read".to_owned(),
4841 why,
4842 seat: Some(seat.clone()),
4843 dispatch_id: None,
4844 };
4845 }
4846 };
4847 let scope = forge_primitives::review::doc_scope_ref(&read.path.to_string_lossy());
4851 let held = surface.reviews(seat.project(), &scope);
4852 let (threads, reviews) = match (held.threads, held.reviews) {
4853 (Ok(threads), Ok(reviews)) => (threads, reviews),
4854 (threads, reviews) => {
4855 let why = threads.err().or(reviews.err()).unwrap_or_default();
4856 return ServerMessage::Error {
4857 what: "document_read".to_owned(),
4858 why: format!("the document's review could not be read: {why}"),
4859 seat: Some(seat.clone()),
4860 dispatch_id: None,
4861 };
4862 }
4863 };
4864 let lines: Option<Vec<String>> = if threads.is_empty() {
4871 None
4872 } else {
4873 std::str::from_utf8(&read.bytes).ok().map(|text| {
4874 let text = text.strip_prefix('\u{feff}').unwrap_or(text);
4878 if text.is_empty() {
4879 vec![String::new()]
4880 } else {
4881 text.lines().map(str::to_owned).collect()
4882 }
4883 })
4884 };
4885 let threads: Vec<ReviewedThreadWire> = surface
4886 .place_document_threads(threads, lines.as_deref())
4887 .into_iter()
4888 .map(|held| ReviewedThreadWire {
4889 thread: held.thread,
4890 resolution: held.placement.map(placement_wire),
4891 })
4892 .collect();
4893 ServerMessage::DocumentRead {
4894 session: seat.clone(),
4895 path: read.path.to_string_lossy().into_owned(),
4896 document_kind: read.kind,
4897 bytes: base64::engine::general_purpose::STANDARD.encode(&read.bytes),
4898 reviewable: read.reviewable,
4899 content_hash: read.content_hash.to_string(),
4900 threads,
4901 reviews,
4902 }
4903}
4904
4905fn refusal_tag(refusal: &DispatchError) -> &'static str {
4912 match refusal {
4913 DispatchError::NoDraftWaiting { .. } => "respond_slack_post",
4916 DispatchError::EmptySlackFeedback { .. } => "empty_slack_feedback",
4919 DispatchError::NoBrowserHandOffWaiting { .. } => "respond_browser_hand_off",
4920 DispatchError::NoTerminalHandOffWaiting { .. } => "respond_terminal_hand_off",
4921 DispatchError::NoSettingsWriteWaiting { .. } => "respond_settings_write",
4922 DispatchError::NoVaultAskWaiting { .. } => "respond_vault_ask",
4923 DispatchError::NoDocumentHandOffWaiting { .. } => "respond_document_hand_off",
4924 _ => "dispatch",
4925 }
4926}
4927
4928struct SentWords {
4938 seat: SessionSlot,
4939 words: String,
4940 landed: fn(&serde_json::Value) -> bool,
4942}
4943
4944fn upsert_landed(body: &serde_json::Value) -> bool {
4946 body.as_bool() == Some(true)
4947}
4948
4949fn submit_landed(body: &serde_json::Value) -> bool {
4951 !body.is_null()
4952}
4953
4954fn words_sent(command: &Command) -> Option<SentWords> {
4955 match command {
4956 Command::UpsertReviewThread { origin: Some(seat), sent_words: Some(words), .. } => {
4957 Some(SentWords { seat: seat.clone(), words: words.clone(), landed: upsert_landed })
4958 }
4959 Command::SubmitReview { origin, summary: Some(words), .. } => {
4960 Some(SentWords { seat: origin.clone(), words: words.clone(), landed: submit_landed })
4961 }
4962 _ => None,
4963 }
4964}
4965
4966fn dispatch_refusal(refusal: &DispatchError, dispatch_id: Option<u64>) -> ServerMessage {
4971 ServerMessage::Error {
4972 what: refusal_tag(refusal).to_owned(),
4973 why: refusal.to_string(),
4974 seat: None,
4975 dispatch_id,
4976 }
4977}
4978
4979fn dispatched(me: u64, command: &Command) {
4984 tracing::debug!(
4985 target: "forge_server::transport",
4986 event_name = "command_dispatched",
4987 client = me,
4988 command = ?command,
4989 "a client command reached the dispatch",
4990 );
4991}
4992
4993fn answers_through_a_reply(command: &Command) -> bool {
5001 matches!(
5002 command,
5003 Command::SpawnWorker { .. }
5004 | Command::DespawnWorker { .. }
5005 | Command::UpsertReviewThread { .. }
5006 | Command::DeleteReviewThread { .. }
5007 | Command::SubmitReview { .. }
5008 | Command::BiometricRelease { .. }
5009 | Command::BiometricEnrol { .. }
5010 | Command::BiometricRewrite { .. }
5011 | Command::VaultEnrolmentMaterial { .. }
5012 )
5013}
5014
5015async fn call_output(
5024 state: &TransportState,
5025 key: &SessionSlot,
5026 call_id: &str,
5027) -> forge_primitives::CallOutput {
5028 use forge_primitives::CallOutput;
5029 let conversations = Arc::clone(&state.conversations);
5030 let key = key.clone();
5031 let logged = key.clone();
5032 let logged_call = call_id.to_owned();
5033 let call_id = call_id.to_owned();
5034 let read = tokio::task::spawn_blocking(move || {
5035 let Some(held) = conversations.get(&key) else {
5036 return CallOutput::NoPath;
5037 };
5038 let conversation = held.lock();
5039 let Some(path) = forge_workspace::output_tail::resolve_call_output_path(
5040 conversation.messages(),
5041 &call_id,
5042 ) else {
5043 return CallOutput::NoPath;
5044 };
5045 match forge_workspace::output_tail::read_output_file_tail(
5046 &path,
5047 forge_workspace::output_tail::CALL_OUTPUT_MAX_LINES,
5048 ) {
5049 None => CallOutput::FileGone,
5050 Some(lines) => CallOutput::Lines(lines),
5051 }
5052 })
5053 .await;
5054 match read {
5055 Ok(output) => output,
5056 Err(error) => {
5057 tracing::warn!(
5061 event_name = "call_output_read_failed",
5062 slot = %logged.display(),
5063 call_id = %logged_call,
5064 %error,
5065 "the call-output read did not finish; the request is answered as no-path",
5066 );
5067 CallOutput::NoPath
5068 }
5069 }
5070}
5071
5072fn answered<T: serde::Serialize + Send + 'static>(
5078 state: &TransportState,
5079 command: Command,
5080 to: u64,
5081 mut answer: oneshot::Receiver<T>,
5082 reply_tx: &mpsc::UnboundedSender<Answer>,
5083) {
5084 let sent = words_sent(&command);
5085 let outcome = state.surface.dispatch(command);
5086 let reply_tx = reply_tx.clone();
5087 if let Ok(reply) = answer.try_recv() {
5094 let body = match outcome {
5095 Ok(()) => serde_json::to_value(&reply).unwrap_or_else(|_| {
5096 serde_json::Value::String("the answer could not be encoded".to_owned())
5097 }),
5098 Err(refusal) => serde_json::Value::String(refusal.to_string()),
5099 };
5100 if let Some(sent) = &sent
5101 && (sent.landed)(&body)
5102 {
5103 Live::lock(&state.live).composer_sent(&sent.seat, &sent.words);
5104 }
5105 let _ = reply_tx.send(Answer::Reply(Box::new(ServerMessage::Reply { reply_to: to, body })));
5106 return;
5107 }
5108 tokio::spawn(async move {
5109 let body = match (outcome, answer.await) {
5110 (Ok(()), Ok(reply)) => serde_json::to_value(reply).unwrap_or_else(|_| {
5111 serde_json::Value::String("the answer could not be encoded".to_owned())
5112 }),
5113 (Err(refusal), _) => serde_json::Value::String(refusal.to_string()),
5114 (Ok(()), Err(_)) => {
5115 serde_json::Value::String("the session closed before answering".to_owned())
5116 }
5117 };
5118 let _ = reply_tx.send(Answer::Reply(Box::new(ServerMessage::Reply { reply_to: to, body })));
5121 });
5122}
5123
5124#[cfg(test)]
5125mod tests {
5126 use std::sync::{Arc, Mutex};
5127
5128 use super::*;
5129 use crate::transport::Heartbeat;
5130 use crate::work::WorkCache;
5131
5132 #[test]
5136 fn every_held_dock_refusal_names_its_own_operation() {
5137 let key = forge_workspace::SessionSlot::lead("Org", "proj");
5138 let id = uuid::Uuid::nil();
5139 assert_eq!(
5140 refusal_tag(&DispatchError::NoDraftWaiting { key: key.clone(), id }),
5141 "respond_slack_post",
5142 );
5143 assert_eq!(
5144 refusal_tag(&DispatchError::NoBrowserHandOffWaiting { key: key.clone(), id }),
5145 "respond_browser_hand_off",
5146 );
5147 assert_eq!(
5148 refusal_tag(&DispatchError::NoSettingsWriteWaiting { key: key.clone(), id }),
5149 "respond_settings_write",
5150 );
5151 assert_eq!(refusal_tag(&DispatchError::NoVaultAskWaiting { key, id }), "respond_vault_ask",);
5152 assert_eq!(refusal_tag(&DispatchError::NoActiveSession), "dispatch");
5153 }
5154
5155 fn a_transport(workspace: Arc<forge_workspace::Workspace>) -> TransportState {
5162 TransportState {
5163 surface: Arc::new(crate::surface::ViewSurface::new(workspace)),
5164 work: Arc::new(WorkCache::new()),
5165 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
5166 live: Mutex::new(crate::live::Live::new()),
5167 settings: crate::transport::settings_channel(),
5168 browser_host: None,
5169 heartbeat: Heartbeat::default(),
5170 }
5171 }
5172
5173 fn a_review_thread(text: &str) -> forge_primitives::review::ReviewThread {
5175 forge_primitives::review::ReviewThread {
5176 id: "t1".to_owned(),
5177 anchor: forge_primitives::review::ReviewAnchor {
5178 path: "src/a.rs".to_owned(),
5179 side: forge_primitives::review::ReviewSide::New,
5180 line: 1,
5181 content_hash: 1,
5182 context: Vec::new(),
5183 base_ref: "main".to_owned(),
5184 },
5185 comments: vec![forge_primitives::review::ReviewComment {
5186 author: forge_primitives::review::ReviewAuthor::User,
5187 text: text.to_owned(),
5188 at: String::new(),
5189 review_id: None,
5190 }],
5191 status: forge_primitives::review::ReviewStatus::Open,
5192 created_at: String::new(),
5193 updated_at: String::new(),
5194 commit: None,
5195 }
5196 }
5197
5198 fn land(transport: &TransportState, seat: &SessionSlot, text: &str) {
5200 Live::lock(&transport.live).apply(&SessionUpdate::DictateEnded {
5201 key: seat.clone(),
5202 outcome: crate::surface::DictateOutcome::Landed {
5203 text: text.to_owned(),
5204 truncated: false,
5205 },
5206 generation: 1,
5207 surface: None,
5208 initiator: None,
5209 });
5210 }
5211
5212 fn still_landed(transport: &TransportState, seat: &SessionSlot) -> bool {
5214 Live::lock(&transport.live).snapshot().composer.landed(seat).is_some()
5215 }
5216
5217 fn upsert(transport: &TransportState, seat: &SessionSlot, words: &str) {
5219 let (tx, rx) = oneshot::channel();
5220 let command = Command::UpsertReviewThread {
5221 project: "proj".to_owned(),
5222 branch: "b".to_owned(),
5223 thread: a_review_thread(words),
5224 origin: Some(seat.clone()),
5225 sent_words: Some(words.to_owned()),
5226 respond: Some(tx),
5227 };
5228 let (reply_tx, replies) = mpsc::unbounded_channel();
5229 answered(transport, command, 7, rx, &reply_tx);
5230 drop(replies);
5231 }
5232
5233 #[tokio::test]
5238 async fn a_review_write_spends_the_landed_words_it_sends() {
5239 let dir = tempfile::tempdir().expect("tempdir");
5240 let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
5241 .expect("the fleet builds");
5242 let transport = a_transport(fleet.workspace());
5243 let seat = SessionSlot::lead("TestOrg", "proj");
5244 land(&transport, &seat, "spoken words");
5245
5246 upsert(&transport, &seat, "something else");
5247 assert!(
5248 still_landed(&transport, &seat),
5249 "words a write does not send are still the seat's to re-hand",
5250 );
5251
5252 upsert(&transport, &seat, "spoken words");
5253 assert!(!still_landed(&transport, &seat), "and the write that sent them spends them");
5254 }
5255
5256 #[tokio::test]
5260 async fn a_review_write_a_closed_store_refuses_leaves_the_words() {
5261 let dir = tempfile::tempdir().expect("tempdir");
5262 let (workspace, _updates) =
5265 forge_workspace::Workspace::testing_stub_with_config_dir(dir.path().to_owned());
5266 let transport = a_transport(workspace);
5267 let seat = SessionSlot::lead("TestOrg", "proj");
5268 land(&transport, &seat, "spoken words");
5269
5270 upsert(&transport, &seat, "spoken words");
5271 assert!(
5272 still_landed(&transport, &seat),
5273 "a store that refused the write did not take the words",
5274 );
5275 }
5276
5277 #[tokio::test]
5283 async fn a_biometric_rewrite_answers_through_its_reply() {
5284 let dir = tempfile::tempdir().expect("tempdir");
5285 let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
5286 .expect("the fleet builds");
5287 let transport = a_transport(fleet.workspace());
5288 let enrolment = uuid::Uuid::new_v4();
5289 let blob = base64::engine::general_purpose::STANDARD.encode(b"sealed");
5290 transport
5291 .surface
5292 .dispatch(Command::BiometricEnrol {
5293 enrolment,
5294 device: "this device".to_owned(),
5295 purpose: "vault".to_owned(),
5296 blob: blob.clone(),
5297 respond: None,
5298 })
5299 .expect("the enrolment lands");
5300 assert!(
5301 answers_through_a_reply(&Command::BiometricRewrite {
5302 enrolment,
5303 purpose: "vault".to_owned(),
5304 blob: blob.clone(),
5305 respond: None,
5306 }),
5307 "the rewrite reports through its reply",
5308 );
5309
5310 let (tx, mut rx) = mpsc::unbounded_channel();
5311 dispatch_answering(
5312 &transport,
5313 Command::BiometricRewrite {
5314 enrolment,
5315 purpose: "vault".to_owned(),
5316 blob: blob.clone(),
5317 respond: None,
5318 },
5319 7,
5320 &tx,
5321 );
5322 let Some(Answer::Reply(message)) = rx.try_recv().ok() else {
5323 panic!("the rewrite's answer is written inside its dispatch");
5324 };
5325 let ServerMessage::Reply { reply_to, body } = *message else {
5326 panic!("the rewrite's answer is not a reply");
5327 };
5328 assert_eq!(reply_to, 7);
5329 assert_eq!(body, serde_json::json!(true), "an existing enrolment is replaced");
5330
5331 dispatch_answering(
5332 &transport,
5333 Command::BiometricRewrite {
5334 enrolment: uuid::Uuid::new_v4(),
5335 purpose: "vault".to_owned(),
5336 blob,
5337 respond: None,
5338 },
5339 8,
5340 &tx,
5341 );
5342 let Some(Answer::Reply(unknown)) = rx.try_recv().ok() else {
5343 panic!("an unknown id still answers");
5344 };
5345 let ServerMessage::Reply { reply_to, body } = *unknown else {
5346 panic!("an unknown id's answer is not a reply");
5347 };
5348 assert_eq!(reply_to, 8);
5349 assert_eq!(body, serde_json::json!(false), "an unknown id cannot be created");
5350 }
5351
5352 #[tokio::test]
5355 async fn a_submit_spends_the_landed_words_its_summary_sends() {
5356 let dir = tempfile::tempdir().expect("tempdir");
5357 let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
5358 .expect("the fleet builds");
5359 let transport = a_transport(fleet.workspace());
5360 let seat = SessionSlot::lead("TestOrg", "proj");
5361 land(&transport, &seat, "spoken words");
5362
5363 let submit = |summary: &str| {
5364 let (tx, rx) = oneshot::channel();
5365 let command = Command::SubmitReview {
5366 project: "proj".to_owned(),
5367 branch: "b".to_owned(),
5368 summary: Some(summary.to_owned()),
5369 thread_ids: Vec::new(),
5370 origin: seat.clone(),
5371 respond: Some(tx),
5372 };
5373 let (reply_tx, replies) = mpsc::unbounded_channel();
5374 answered(&transport, command, 7, rx, &reply_tx);
5375 drop(replies);
5376 };
5377
5378 submit("a different overview");
5379 assert!(still_landed(&transport, &seat), "a summary that is not the words leaves them");
5380
5381 submit("spoken words");
5382 assert!(!still_landed(&transport, &seat), "and the summary that is them spends them");
5383 }
5384
5385 #[tokio::test]
5389 async fn a_call_output_read_resolves_from_the_transports_held_copy() {
5390 let dir = tempfile::tempdir().expect("tempdir");
5391 let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
5392 .expect("the fleet builds");
5393 let conversations = Arc::new(crate::transport::conversation::Conversations::new());
5394 let state = TransportState {
5395 surface: fleet.surface(),
5396 work: Arc::new(WorkCache::new()),
5397 conversations: Arc::clone(&conversations),
5398 live: Mutex::new(crate::live::Live::new()),
5399 settings: crate::transport::settings_channel(),
5400 browser_host: None,
5401 heartbeat: Heartbeat::default(),
5402 };
5403 let seat = SessionSlot::lead("TestOrg", "proj");
5404 conversations
5405 .insert(&seat, crate::transport::conversation::Conversation::new(Vec::new(), 0, 0));
5406
5407 let path = std::env::temp_dir()
5408 .join(format!("forge-call-output-transport-{}.log", std::process::id()));
5409 std::fs::write(&path, "one\ntwo\n").expect("write the output file");
5410 let msg: forge_primitives::Message = serde_json::from_value(serde_json::json!({
5411 "type": "system",
5412 "subtype": "task_notification",
5413 "task_id": "t-1",
5414 "tool_use_id": "tu-1",
5415 "status": "completed",
5416 "output_file": path.display().to_string(),
5417 "summary": "done",
5418 "session_id": "sess-1",
5419 "uuid": "u-1",
5420 }))
5421 .expect("a task notification frame");
5422 conversations
5423 .apply(&SessionUpdate::ChatAppended { key: seat.clone(), msg, source: None }, Some(1));
5424
5425 assert_eq!(
5426 call_output(&state, &seat, "tu-1").await,
5427 forge_primitives::CallOutput::Lines(vec!["one".to_owned(), "two".to_owned()]),
5428 "the copy's own frame names the file the read returns",
5429 );
5430 assert_eq!(
5431 call_output(&state, &seat, "tu-unknown").await,
5432 forge_primitives::CallOutput::NoPath,
5433 "a call the copy does not name is the no-path reason",
5434 );
5435 std::fs::remove_file(&path).expect("remove the output file");
5436 assert_eq!(
5437 call_output(&state, &seat, "tu-1").await,
5438 forge_primitives::CallOutput::FileGone,
5439 "a named file that has gone is its own reason, not an empty tail",
5440 );
5441 }
5442
5443 #[tokio::test]
5449 async fn a_panicking_read_answers_the_loop_and_a_cancelled_one_does_not() {
5450 let (reply, mut answers) = mpsc::unbounded_channel();
5451 let guard = ReadAnswerGuard {
5454 reply,
5455 subject: Subject::Home,
5456 seat: None,
5457 peer: "127.0.0.1:1".parse().expect("a peer"),
5458 open_at: Instant::now(),
5459 queue_ms: 0.0,
5460 prelude_ms: 0.0,
5461 queued_ms: 0.0,
5462 hold_ms: 0.0,
5463 answered: false,
5464 };
5465 let _task = spawn_read(guard, async { panic!("the encode broke") });
5466 let answer = tokio::time::timeout(Duration::from_secs(5), answers.recv())
5467 .await
5468 .expect("the panic answered the loop")
5469 .expect("the channel is open");
5470 let Answer::Read(read) = answer else { panic!("a read's answer, not a reply") };
5471 assert!(
5472 read.outcome
5473 .as_ref()
5474 .err()
5475 .is_some_and(|why| why.to_string().contains("without an answer")),
5476 "the refusal names what happened",
5477 );
5478
5479 let (reply, mut answers) = mpsc::unbounded_channel();
5481 let cancelled = ReadAnswerGuard {
5482 reply,
5483 subject: Subject::Home,
5484 seat: None,
5485 peer: "127.0.0.1:1".parse().expect("a peer"),
5486 open_at: Instant::now(),
5487 queue_ms: 0.0,
5488 prelude_ms: 0.0,
5489 queued_ms: 0.0,
5490 hold_ms: 0.0,
5491 answered: false,
5492 };
5493 drop(cancelled);
5494 assert!(answers.try_recv().is_err(), "a cancelled read answered its own ask twice");
5495 }
5496
5497 #[tokio::test]
5502 async fn a_ping_is_the_only_thing_that_can_go_unanswered() {
5503 let dead = Duration::from_secs(60);
5504 let keepalive = Keepalive::default();
5505 let start = Instant::now();
5506
5507 assert_eq!(keepalive.beat(start, dead), Beat::Ping, "no ping out: ask");
5508 assert_eq!(
5509 keepalive.beat(start + Duration::from_secs(65), dead),
5510 Beat::Ping,
5511 "nothing was sent, so nothing went unanswered and the beat asks",
5512 );
5513
5514 let mut sink = NullSink;
5516 keepalive.ping(&mut sink).await.expect("the ping goes");
5517 let at = keepalive.sent().expect("the send stamped the deadline");
5518 assert_eq!(
5519 keepalive.beat(at + Duration::from_secs(30), dead),
5520 Beat::Wait,
5521 "inside the bound the answer is still owed",
5522 );
5523 assert_eq!(
5524 keepalive.beat(at + Duration::from_secs(65), dead),
5525 Beat::Reap,
5526 "past it the peer has not answered, and that is the evidence",
5527 );
5528 }
5529
5530 #[test]
5535 fn asking_does_not_stamp_the_deadline() {
5536 let dead = Duration::from_secs(60);
5537 let keepalive = Keepalive::default();
5538 let start = Instant::now();
5539
5540 assert_eq!(keepalive.beat(start, dead), Beat::Ping);
5541 assert_eq!(keepalive.sent(), None, "the deadline starts when the ping is on the wire");
5542 }
5543
5544 #[test]
5549 fn only_discrete_gestures_are_recorded() {
5550 use forge_primitives::browser::InputModifiers;
5551
5552 let mouse = |action| BrowserInputEvent::Mouse {
5553 action,
5554 x: 1.0,
5555 y: 2.0,
5556 button: 0,
5557 clicks: 1,
5558 modifiers: InputModifiers::default(),
5559 };
5560 assert!(discrete_gesture(&mouse(MouseAction::Press)));
5561 assert!(discrete_gesture(&mouse(MouseAction::Release)));
5562 assert!(!discrete_gesture(&mouse(MouseAction::Move)));
5563 assert!(!discrete_gesture(&BrowserInputEvent::Wheel {
5564 x: 1.0,
5565 y: 2.0,
5566 delta_x: 3.0,
5567 delta_y: 4.0,
5568 modifiers: InputModifiers::default(),
5569 }));
5570 assert!(discrete_gesture(&BrowserInputEvent::Text { text: "x".to_owned() }));
5571 assert!(discrete_gesture(&BrowserInputEvent::Key {
5572 down: true,
5573 key: "Enter".to_owned(),
5574 code: "Enter".to_owned(),
5575 windows_vk: 13,
5576 modifiers: InputModifiers::default(),
5577 }));
5578 assert!(discrete_gesture(&BrowserInputEvent::Key {
5579 down: false,
5580 key: "Enter".to_owned(),
5581 code: "Enter".to_owned(),
5582 windows_vk: 13,
5583 modifiers: InputModifiers::default(),
5584 }));
5585 }
5586
5587 #[test]
5590 fn every_gesture_kind_names_itself() {
5591 use forge_primitives::browser::InputModifiers;
5592
5593 let mouse = |action| BrowserInputEvent::Mouse {
5594 action,
5595 x: 1.0,
5596 y: 2.0,
5597 button: 0,
5598 clicks: 1,
5599 modifiers: InputModifiers::default(),
5600 };
5601 assert_eq!(gesture_kind(&mouse(MouseAction::Press)), "mouse press");
5602 assert_eq!(gesture_kind(&mouse(MouseAction::Release)), "mouse release");
5603 assert_eq!(gesture_kind(&mouse(MouseAction::Move)), "mouse move");
5604 assert_eq!(
5605 gesture_kind(&BrowserInputEvent::Wheel {
5606 x: 1.0,
5607 y: 2.0,
5608 delta_x: 3.0,
5609 delta_y: 4.0,
5610 modifiers: InputModifiers::default(),
5611 }),
5612 "wheel",
5613 );
5614 let key = |down| BrowserInputEvent::Key {
5615 down,
5616 key: "Enter".to_owned(),
5617 code: "Enter".to_owned(),
5618 windows_vk: 13,
5619 modifiers: InputModifiers::default(),
5620 };
5621 assert_eq!(gesture_kind(&key(true)), "key down");
5622 assert_eq!(gesture_kind(&key(false)), "key up");
5623 assert_eq!(gesture_kind(&BrowserInputEvent::Text { text: "x".to_owned() }), "text");
5624 }
5625
5626 #[test]
5629 fn a_gesture_is_named_by_its_kind_not_its_text() {
5630 assert_eq!(gesture_kind(&BrowserInputEvent::Text { text: "hunter2".to_owned() }), "text");
5631 assert_eq!(text_chars(&BrowserInputEvent::Text { text: "hunter2".to_owned() }), Some(7));
5632 assert_eq!(
5633 text_chars(&BrowserInputEvent::Key {
5634 down: false,
5635 key: "a".to_owned(),
5636 code: "KeyA".to_owned(),
5637 windows_vk: 65,
5638 modifiers: forge_primitives::browser::InputModifiers::default(),
5639 }),
5640 None
5641 );
5642 }
5643
5644 fn attached_from() -> std::net::SocketAddr {
5647 "127.0.0.1:0".parse().expect("a test address")
5648 }
5649
5650 struct NullSink;
5653
5654 impl futures_util::Sink<Message> for NullSink {
5655 type Error = axum::Error;
5656 fn poll_ready(
5657 self: std::pin::Pin<&mut Self>,
5658 _cx: &mut std::task::Context<'_>,
5659 ) -> std::task::Poll<Result<(), Self::Error>> {
5660 std::task::Poll::Ready(Ok(()))
5661 }
5662 fn start_send(self: std::pin::Pin<&mut Self>, _item: Message) -> Result<(), Self::Error> {
5663 Ok(())
5664 }
5665 fn poll_flush(
5666 self: std::pin::Pin<&mut Self>,
5667 _cx: &mut std::task::Context<'_>,
5668 ) -> std::task::Poll<Result<(), Self::Error>> {
5669 std::task::Poll::Ready(Ok(()))
5670 }
5671 fn poll_close(
5672 self: std::pin::Pin<&mut Self>,
5673 _cx: &mut std::task::Context<'_>,
5674 ) -> std::task::Poll<Result<(), Self::Error>> {
5675 std::task::Poll::Ready(Ok(()))
5676 }
5677 }
5678
5679 struct YieldsOnce {
5682 yielded: bool,
5683 }
5684
5685 impl futures_util::Sink<Message> for YieldsOnce {
5686 type Error = axum::Error;
5687
5688 fn poll_ready(
5689 self: std::pin::Pin<&mut Self>,
5690 _cx: &mut std::task::Context<'_>,
5691 ) -> std::task::Poll<Result<(), Self::Error>> {
5692 std::task::Poll::Ready(Ok(()))
5693 }
5694 fn start_send(self: std::pin::Pin<&mut Self>, _item: Message) -> Result<(), Self::Error> {
5695 Ok(())
5696 }
5697 fn poll_flush(
5698 mut self: std::pin::Pin<&mut Self>,
5699 cx: &mut std::task::Context<'_>,
5700 ) -> std::task::Poll<Result<(), Self::Error>> {
5701 if !self.yielded {
5702 self.yielded = true;
5703 cx.waker().wake_by_ref();
5704 return std::task::Poll::Pending;
5705 }
5706 std::task::Poll::Ready(Ok(()))
5707 }
5708 fn poll_close(
5709 self: std::pin::Pin<&mut Self>,
5710 _cx: &mut std::task::Context<'_>,
5711 ) -> std::task::Poll<Result<(), Self::Error>> {
5712 std::task::Poll::Ready(Ok(()))
5713 }
5714 }
5715
5716 struct Recorded(Arc<Mutex<Vec<Message>>>);
5719
5720 impl futures_util::Sink<Message> for Recorded {
5721 type Error = axum::Error;
5722
5723 fn poll_ready(
5724 self: std::pin::Pin<&mut Self>,
5725 _cx: &mut std::task::Context<'_>,
5726 ) -> std::task::Poll<Result<(), Self::Error>> {
5727 std::task::Poll::Ready(Ok(()))
5728 }
5729 fn start_send(self: std::pin::Pin<&mut Self>, item: Message) -> Result<(), Self::Error> {
5730 if let Ok(mut held) = self.0.lock() {
5731 held.push(item);
5732 }
5733 Ok(())
5734 }
5735 fn poll_flush(
5736 self: std::pin::Pin<&mut Self>,
5737 _cx: &mut std::task::Context<'_>,
5738 ) -> std::task::Poll<Result<(), Self::Error>> {
5739 std::task::Poll::Ready(Ok(()))
5740 }
5741 fn poll_close(
5742 self: std::pin::Pin<&mut Self>,
5743 _cx: &mut std::task::Context<'_>,
5744 ) -> std::task::Poll<Result<(), Self::Error>> {
5745 std::task::Poll::Ready(Ok(()))
5746 }
5747 }
5748
5749 #[tokio::test]
5754 async fn a_batched_text_frame_crosses_the_encoder() {
5755 let recorded = Arc::new(Mutex::new(Vec::new()));
5756 let mut socket = ClientSocket::new(Recorded(Arc::clone(&recorded)), true);
5757 let text =
5758 format!("{{\"kind\":\"update\",\"pad\":\"{}\"}}", "x".repeat(COMPRESSED_MIN_BYTES));
5759 futures_util::SinkExt::send(&mut socket, Message::Text(text.clone().into()))
5760 .await
5761 .expect("the frame goes");
5762 let held = recorded.lock().expect("the frame log");
5763 let Some(Message::Binary(bytes)) = held.first() else {
5764 panic!("a text frame over the threshold crossed as text: {:?}", held.first())
5765 };
5766 assert_eq!(bytes.first().copied(), Some(COMPRESSED_JSON_TAG), "the compressed kind tag");
5767 let mut body = String::new();
5768 std::io::Read::read_to_string(&mut flate2::read::GzDecoder::new(&bytes[1..]), &mut body)
5769 .expect("the body inflates");
5770 assert_eq!(body, text, "the body is the frame it stood for");
5771 }
5772
5773 #[tokio::test]
5779 async fn a_text_frame_the_encoder_leaves_alone_is_not_rebuilt() {
5780 let recorded = Arc::new(Mutex::new(Vec::new()));
5781 let mut socket = ClientSocket::new(Recorded(Arc::clone(&recorded)), true);
5782 let frame = Message::Text("{\"kind\":\"update\"}".into());
5783 let Message::Text(kept) = &frame else { panic!("a text frame") };
5784 let kept = kept.clone();
5787 let address = kept.as_str().as_ptr();
5788 futures_util::SinkExt::send(&mut socket, frame).await.expect("the frame goes");
5789 {
5790 let held = recorded.lock().expect("the frame log");
5791 let Some(Message::Text(sent)) = held.first() else {
5792 panic!("an under-threshold frame was not text: {:?}", held.first())
5793 };
5794 assert_eq!(
5795 sent.as_str().as_ptr(),
5796 address,
5797 "the encoder rebuilt a text frame it should have passed through"
5798 );
5799 }
5800
5801 let recorded = Arc::new(Mutex::new(Vec::new()));
5804 let mut socket = ClientSocket::new(Recorded(Arc::clone(&recorded)), false);
5805 let frame = Message::Text("x".repeat(COMPRESSED_MIN_BYTES).into());
5806 let Message::Text(kept) = &frame else { panic!("a text frame") };
5807 let kept = kept.clone();
5808 let address = kept.as_str().as_ptr();
5809 futures_util::SinkExt::send(&mut socket, frame).await.expect("the frame goes");
5810 {
5811 let held = recorded.lock().expect("the frame log");
5812 let Some(Message::Text(sent)) = held.first() else {
5813 panic!("a non-declaring client's frame was not text: {:?}", held.first())
5814 };
5815 assert_eq!(
5816 sent.as_str().as_ptr(),
5817 address,
5818 "a non-declaring client's frame was rebuilt"
5819 );
5820 }
5821
5822 let recorded = Arc::new(Mutex::new(Vec::new()));
5825 let mut socket = ClientSocket::new(Recorded(Arc::clone(&recorded)), true);
5826 let frame = Message::Text("x".repeat(COMPRESSED_MIN_BYTES).into());
5827 futures_util::SinkExt::send(&mut socket, frame).await.expect("the frame goes");
5828 let held = recorded.lock().expect("the frame log");
5829 assert!(
5830 matches!(
5831 held.first(),
5832 Some(Message::Binary(bytes)) if bytes.first().copied() == Some(COMPRESSED_JSON_TAG)
5833 ),
5834 "a frame at the threshold must cross compressed",
5835 );
5836 }
5837
5838 struct FakeClock {
5842 starts: std::cell::RefCell<Vec<usize>>,
5843 base: Instant,
5844 }
5845
5846 impl WriteClock for FakeClock {
5847 fn start(&self, frame: &Message) -> Instant {
5848 self.starts.borrow_mut().push(frame_bytes(frame));
5849 self.base
5850 }
5851
5852 fn end(&self) -> Instant {
5853 self.base + Duration::from_secs(1)
5854 }
5855 }
5856
5857 struct HeldSink {
5859 ready: std::sync::Arc<std::sync::atomic::AtomicBool>,
5860 }
5861
5862 impl futures_util::Sink<Message> for HeldSink {
5863 type Error = axum::Error;
5864 fn poll_ready(
5865 self: std::pin::Pin<&mut Self>,
5866 _cx: &mut std::task::Context<'_>,
5867 ) -> std::task::Poll<Result<(), Self::Error>> {
5868 if self.ready.load(std::sync::atomic::Ordering::Relaxed) {
5869 std::task::Poll::Ready(Ok(()))
5870 } else {
5871 std::task::Poll::Pending
5872 }
5873 }
5874 fn start_send(self: std::pin::Pin<&mut Self>, _item: Message) -> Result<(), Self::Error> {
5875 Ok(())
5876 }
5877 fn poll_flush(
5878 self: std::pin::Pin<&mut Self>,
5879 _cx: &mut std::task::Context<'_>,
5880 ) -> std::task::Poll<Result<(), Self::Error>> {
5881 std::task::Poll::Ready(Ok(()))
5882 }
5883 fn poll_close(
5884 self: std::pin::Pin<&mut Self>,
5885 _cx: &mut std::task::Context<'_>,
5886 ) -> std::task::Poll<Result<(), Self::Error>> {
5887 std::task::Poll::Ready(Ok(()))
5888 }
5889 }
5890
5891 struct StoppedReading {
5896 inbound: mpsc::UnboundedReceiver<Message>,
5897 writing: Option<oneshot::Sender<()>>,
5900 }
5901
5902 impl futures_util::Stream for StoppedReading {
5903 type Item = Result<Message, axum::Error>;
5904
5905 fn poll_next(
5906 mut self: std::pin::Pin<&mut Self>,
5907 cx: &mut std::task::Context<'_>,
5908 ) -> std::task::Poll<Option<Self::Item>> {
5909 self.inbound.poll_recv(cx).map(|said| said.map(Ok))
5910 }
5911 }
5912
5913 impl futures_util::Sink<Message> for StoppedReading {
5914 type Error = axum::Error;
5915
5916 fn poll_ready(
5917 self: std::pin::Pin<&mut Self>,
5918 _cx: &mut std::task::Context<'_>,
5919 ) -> std::task::Poll<Result<(), Self::Error>> {
5920 std::task::Poll::Ready(Ok(()))
5921 }
5922 fn start_send(self: std::pin::Pin<&mut Self>, _item: Message) -> Result<(), Self::Error> {
5923 Ok(())
5924 }
5925 fn poll_flush(
5926 mut self: std::pin::Pin<&mut Self>,
5927 _cx: &mut std::task::Context<'_>,
5928 ) -> std::task::Poll<Result<(), Self::Error>> {
5929 if let Some(writing) = self.writing.take() {
5930 let _ = writing.send(());
5931 }
5932 std::task::Poll::Pending
5933 }
5934 fn poll_close(
5935 self: std::pin::Pin<&mut Self>,
5936 _cx: &mut std::task::Context<'_>,
5937 ) -> std::task::Poll<Result<(), Self::Error>> {
5938 std::task::Poll::Ready(Ok(()))
5939 }
5940 }
5941
5942 #[tokio::test]
5947 async fn the_deadline_starts_when_the_ping_goes_out() {
5948 let ready = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
5949 let mut sink = HeldSink { ready: std::sync::Arc::clone(&ready) };
5950 let keepalive = Keepalive::default();
5951 let mut sending = std::pin::pin!(keepalive.ping(&mut sink));
5952
5953 assert!(
5954 futures_util::poll!(sending.as_mut()).is_pending(),
5955 "the write is held in the buffer",
5956 );
5957 assert_eq!(
5958 keepalive.sent(),
5959 None,
5960 "nothing may be owed while the ping is still to be written",
5961 );
5962
5963 ready.store(true, std::sync::atomic::Ordering::Relaxed);
5964 assert!(futures_util::poll!(sending.as_mut()).is_ready(), "the write completed");
5965 assert!(keepalive.sent().is_some(), "and the answer is owed from the send's return");
5966 }
5967
5968 #[tokio::test]
5971 async fn an_answer_clears_the_outstanding_ping() {
5972 let dead = Duration::from_secs(60);
5973 let keepalive = Keepalive::default();
5974 let mut sink = NullSink;
5975 keepalive.ping(&mut sink).await.expect("the ping goes");
5976 let at = keepalive.sent().expect("the send stamped the deadline");
5977
5978 keepalive.heard();
5979
5980 assert_eq!(
5981 keepalive.beat(at + Duration::from_secs(65), dead),
5982 Beat::Ping,
5983 "the answer retired the ping, so the next beat asks again",
5984 );
5985 }
5986
5987 #[tokio::test(start_paused = true)]
5997 async fn a_peer_that_stops_reading_is_cut_at_the_write_bound() {
5998 let fleet =
5999 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
6000 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
6001
6002 let (inbound, peer_in) = mpsc::unbounded_channel();
6003 let (writing, stuck) = oneshot::channel();
6004 let mut peer =
6005 ClientSocket::new(StoppedReading { inbound: peer_in, writing: Some(writing) }, false);
6006 let settings = crate::transport::settings_channel();
6007 let driven = tokio::spawn({
6008 let state = Arc::clone(&state);
6009 async move {
6010 let receiver = settings.subscribe();
6011 drive(
6012 &mut peer,
6013 &state,
6014 attached_from(),
6015 receiver,
6016 None,
6017 &mut ConnectionLife::new(),
6018 )
6019 .await
6020 }
6021 });
6022
6023 inbound
6026 .send(Message::Text(
6027 serde_json::to_string(&ClientMessage::Subscribe {
6028 what: Subject::Usage,
6029 answering: false,
6030 review: false,
6031 })
6032 .expect("the subscribe encodes")
6033 .into(),
6034 ))
6035 .expect("the peer's one message goes");
6036 stuck.await.expect("the subscribe's answer reaches the socket");
6037
6038 let outcome = tokio::time::timeout(WRITE_BOUND * 2, driven)
6041 .await
6042 .expect("the connection ends at the write bound rather than never")
6043 .expect("the connection task joins");
6044 let error = outcome.expect_err("a peer that stopped reading must not end cleanly");
6045 let stalled = error.downcast_ref::<WriteStalled>().unwrap_or_else(|| {
6046 panic!(
6047 "the connection ends on the write bound rather than on a socket error: {error:?}"
6048 )
6049 });
6050 assert!(
6051 stalled.waited >= WRITE_BOUND,
6052 "the bound's own record carries how long the write ran: {:?}",
6053 stalled.waited,
6054 );
6055 }
6056
6057 #[tokio::test(start_paused = true)]
6062 async fn a_ping_that_cannot_be_written_ends_at_the_write_bound() {
6063 let ready = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
6064 let mut sink = HeldSink { ready: std::sync::Arc::clone(&ready) };
6065 let keepalive = Keepalive::default();
6066
6067 let outcome = tokio::time::timeout(WRITE_BOUND * 2, keepalive.ping(&mut sink))
6068 .await
6069 .expect("the ping ends at the write bound rather than never");
6070 let error = outcome.expect_err("a ping the buffer cannot take is not sent");
6071
6072 assert!(
6073 error.downcast_ref::<WriteStalled>().is_some(),
6074 "the ping ends on the write bound rather than on a sink error: {error:?}",
6075 );
6076 assert_eq!(
6077 keepalive.sent(),
6078 None,
6079 "and no deadline is stamped for a ping the peer never saw",
6080 );
6081 }
6082
6083 #[tokio::test(start_paused = true)]
6087 async fn a_greeting_that_cannot_be_written_ends_at_the_write_bound() {
6088 let fleet =
6089 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
6090 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
6091 let (_keep, inbound) = mpsc::unbounded_channel();
6092 let mut peer = StoppedReading { inbound, writing: None };
6093
6094 tokio::time::timeout(
6095 WRITE_BOUND * 2,
6096 greet(ClientSocket::new(&mut peer, false), state, attached_from(), false),
6097 )
6098 .await
6099 .expect("the greeting ends at the write bound rather than never");
6100 }
6101
6102 fn a_frame() -> ServerMessage {
6105 ServerMessage::Error {
6106 what: "frame".to_owned(),
6107 why: "a measured write".to_owned(),
6108 seat: None,
6109 dispatch_id: None,
6110 }
6111 }
6112
6113 #[tokio::test]
6118 async fn the_write_is_timed_from_after_the_frame_is_serialised() {
6119 let bytes = serde_json::to_string(&a_frame()).expect("the frame encodes").len();
6120 let clock = FakeClock { starts: std::cell::RefCell::new(Vec::new()), base: Instant::now() };
6121
6122 let write = send_frame_timed(&mut ClientSocket::new(NullSink, false), a_frame(), &clock)
6123 .await
6124 .expect("the write lands");
6125 assert!(!write.waited, "a sink that never yields is reported as not waiting");
6126 assert_eq!(write.bytes, bytes, "the record carries the frame's own size");
6127 assert_eq!(write.write_ms, 1000.0, "the record's elapsed is the clock's, exactly");
6128 assert_eq!(
6129 clock.starts.borrow().as_slice(),
6130 &[bytes],
6131 "the clock is read once, with the serialised frame",
6132 );
6133 }
6134
6135 #[tokio::test]
6141 async fn a_compressed_write_reports_the_bytes_it_wrote() {
6142 let large = ServerMessage::Snapshot {
6143 subject: Subject::Home,
6144 data: json!({ "padding": "x".repeat(100_000) }),
6145 };
6146 let text = serde_json::to_string(&large).expect("the frame encodes").len();
6147 let clock = FakeClock { starts: std::cell::RefCell::new(Vec::new()), base: Instant::now() };
6148 let mut socket = ClientSocket::new(NullSink, true);
6149
6150 let write = send_frame_timed(&mut socket, large, &clock).await.expect("the write lands");
6151 assert!(
6152 write.bytes < text / 4,
6153 "the record carries the compressed size, not the text's: {} of {text}",
6154 write.bytes,
6155 );
6156 assert_eq!(
6157 clock.starts.borrow().as_slice(),
6158 &[write.bytes],
6159 "the clock is handed the frame actually written",
6160 );
6161
6162 let small = serde_json::to_string(&a_frame()).expect("the frame encodes").len();
6163 let write =
6164 send_frame_timed(&mut socket, a_frame(), &clock).await.expect("the write lands");
6165 assert_eq!(write.bytes, small, "a frame under the threshold stays the text frame it was");
6166 }
6167
6168 #[test]
6173 fn the_compression_threshold_is_exact() {
6174 let at_len = |len: usize| {
6175 let base = serde_json::to_string(&ServerMessage::Snapshot {
6176 subject: Subject::Home,
6177 data: json!({ "pad": "" }),
6178 })
6179 .expect("the frame encodes")
6180 .len();
6181 let pad = "x".repeat(len - base);
6182 let text = serde_json::to_string(&ServerMessage::Snapshot {
6183 subject: Subject::Home,
6184 data: json!({ "pad": pad }),
6185 })
6186 .expect("the frame encodes");
6187 assert_eq!(text.len(), len, "precondition: the frame is exactly {len} bytes");
6188 text
6189 };
6190
6191 let under = at_len(COMPRESSED_MIN_BYTES - 1);
6192 assert!(
6193 matches!(frame_for_flag(true, &under), Message::Text(_)),
6194 "a byte under the threshold stayed not-text",
6195 );
6196 let at = at_len(COMPRESSED_MIN_BYTES);
6197 assert!(
6198 matches!(frame_for_flag(true, &at), Message::Binary(_)),
6199 "a frame at exactly the threshold crossed as text",
6200 );
6201 }
6202
6203 #[tokio::test]
6206 async fn a_write_that_yields_is_timed_and_says_so() {
6207 let bytes = serde_json::to_string(&a_frame()).expect("the frame encodes").len();
6208 let clock = FakeClock { starts: std::cell::RefCell::new(Vec::new()), base: Instant::now() };
6209
6210 let mut yielding = ClientSocket::new(YieldsOnce { yielded: false }, false);
6211 let write =
6212 send_frame_timed(&mut yielding, a_frame(), &clock).await.expect("the write lands");
6213 assert!(write.waited, "a flush that yielded once is reported as having waited");
6214 assert_eq!(write.bytes, bytes, "the record carries the frame's own size");
6215 assert_eq!(write.write_ms, 1000.0, "the record's elapsed is the clock's, exactly");
6216 }
6217
6218 fn boundary_of(
6221 window: &crate::transport::conversation::Conversation,
6222 ) -> Option<forge_primitives::TranscriptAnchor> {
6223 super::boundary_candidates(window).into_iter().next()
6224 }
6225
6226 #[tokio::test]
6234 async fn an_attached_connection_is_what_a_hand_off_claims_its_open_on() {
6235 let dir = tempfile::tempdir().expect("tempdir");
6236 let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
6237 .expect("the fleet builds");
6238 let root = dir.path().join("proj");
6239 std::fs::create_dir_all(root.join("docs")).expect("the project tree");
6240 std::fs::write(root.join("docs/readme.md"), b"# hi\n").expect("the document");
6241
6242 let state = TransportState {
6243 surface: fleet.surface(),
6244 work: Arc::new(WorkCache::new()),
6245 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
6246 live: Mutex::new(crate::live::Live::new()),
6247 settings: crate::transport::settings_channel(),
6248 browser_host: None,
6249 heartbeat: Heartbeat::default(),
6250 };
6251 let mut updates = state.surface.subscribe_client(true);
6255 let seat = SessionSlot::lead("TestOrg", "proj");
6256
6257 let refused = fleet
6258 .workspace()
6259 .hand_off_document(&seat, "docs/readme.md", false)
6260 .await
6261 .expect_err("no connection, so nothing to open on");
6262 assert_eq!(refused, forge_workspace::documents::NO_VIEWER_CLIENT);
6263
6264 let attached = AttachedClient::new(&state.surface);
6265 let handed = {
6266 let ws = fleet.workspace();
6267 let seat = seat.clone();
6268 tokio::spawn(async move { ws.hand_off_document(&seat, "docs/readme.md", false).await })
6269 };
6270 let id = tokio::time::timeout(std::time::Duration::from_secs(5), async {
6271 loop {
6272 match updates
6273 .recv()
6274 .await
6275 .map(|frame| std::sync::Arc::unwrap_or_clone(frame.update))
6276 {
6277 Some(SessionUpdate::DocumentHandOffPending { id, .. }) => break id,
6278 Some(_) => {}
6279 None => panic!("the stream ended before the ask"),
6280 }
6281 }
6282 })
6283 .await
6284 .expect("an attached connection is what the hand-off parks on");
6285
6286 state
6289 .surface
6290 .dispatch(forge_workspace::Command::AnswerDocumentHandOff {
6291 key: seat.clone(),
6292 id,
6293 answer: forge_primitives::document::DocumentHandOffEnding::Approve {
6294 review_id: None,
6295 note: None,
6296 },
6297 })
6298 .expect("the answer lands on the parked ask");
6299 handed.await.expect("the task joins").expect("and the park releases");
6300
6301 drop(attached);
6302 let refused = fleet
6303 .workspace()
6304 .hand_off_document(&seat, "docs/readme.md", false)
6305 .await
6306 .expect_err("the connection went, so the claim goes with it");
6307 assert_eq!(refused, forge_workspace::documents::NO_VIEWER_CLIENT);
6308 }
6309
6310 #[tokio::test]
6316 async fn escalating_hands_back_what_the_replaced_stream_had_queued() {
6317 let fleet =
6318 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
6319 let state = TransportState {
6320 surface: fleet.surface(),
6321 work: Arc::new(WorkCache::new()),
6322 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
6323 live: Mutex::new(crate::live::Live::new()),
6324 settings: crate::transport::settings_channel(),
6325 browser_host: None,
6326 heartbeat: Heartbeat::default(),
6327 };
6328
6329 let mut updates = Some((state.surface.subscribe_client(false), false));
6331 fleet.emit(SessionUpdate::CatalogLoaded);
6332 fleet.emit(SessionUpdate::CatalogLoaded);
6333
6334 let handed_back = open_stream(&state, &mut updates, true);
6335
6336 assert_eq!(
6337 handed_back.len(),
6338 2,
6339 "both updates the replaced stream was holding come back to be forwarded",
6340 );
6341 assert!(
6342 matches!(updates, Some((_, true))),
6343 "and the connection is on an answering stream now",
6344 );
6345 }
6346
6347 #[tokio::test]
6352 async fn a_queued_frame_reaches_the_client_capped() {
6353 let fleet =
6354 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
6355 let state = TransportState {
6356 surface: fleet.surface(),
6357 work: Arc::new(WorkCache::new()),
6358 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
6359 live: Mutex::new(crate::live::Live::new()),
6360 settings: crate::transport::settings_channel(),
6361 browser_host: None,
6362 heartbeat: Heartbeat::default(),
6363 };
6364 let seat = forge_primitives::SessionSlot::lead("TestOrg", "proj");
6365 let msg: forge_primitives::Message = serde_json::from_value(serde_json::json!({
6366 "type": "user",
6367 "uuid": "tool-result-1",
6368 "session_id": "s",
6369 "message": {"role": "user", "content": "the call"},
6370 "tool_use_result": {"payload": "x".repeat(100_000)},
6371 }))
6372 .expect("a user frame");
6373 let frame = SeatUpdate {
6374 seq: Some(9),
6375 update: Arc::new(SessionUpdate::ChatAppended { key: seat.clone(), source: None, msg }),
6376 };
6377
6378 let queued = forward_queued(&state, vec![frame], &[Subject::Session(seat)], 1);
6379
6380 assert_eq!(queued.len(), 1, "the frame is forwarded to the seat's watcher");
6381 let SessionUpdate::ChatAppended {
6382 msg: forge_primitives::Message::User { tool_use_result, .. },
6383 ..
6384 } = &*queued[0].update
6385 else {
6386 panic!("the queued frame is the chat frame it was queued as")
6387 };
6388 let record = tool_use_result.as_ref().expect("the frame carries its record");
6389 assert_eq!(record["omitted"], true, "the queued record is bounded: {record}");
6390 }
6391
6392 #[tokio::test]
6396 async fn a_later_observing_subscribe_does_not_downgrade_the_stream() {
6397 let fleet =
6398 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
6399 let state = TransportState {
6400 surface: fleet.surface(),
6401 work: Arc::new(WorkCache::new()),
6402 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
6403 live: Mutex::new(crate::live::Live::new()),
6404 settings: crate::transport::settings_channel(),
6405 browser_host: None,
6406 heartbeat: Heartbeat::default(),
6407 };
6408
6409 let mut updates = Some((state.surface.subscribe_client(true), true));
6410 fleet.emit(SessionUpdate::CatalogLoaded);
6411
6412 let handed_back = open_stream(&state, &mut updates, false);
6413
6414 assert!(handed_back.is_empty(), "the stream was not replaced, so nothing is handed back");
6415 assert!(
6416 matches!(updates, Some((_, true))),
6417 "and the client is still counted as one that can answer",
6418 );
6419 }
6420
6421 #[test]
6426 fn an_answer_is_stamped_with_the_connection_that_sent_it() {
6427 let seat = SessionSlot::lead("TestOrg", "proj");
6428 let id = uuid::Uuid::from_u128(7);
6429 let answer = |client| forge_primitives::browser::HandOffAnswer {
6430 ending: forge_primitives::browser::HandOffEnding::Feedback,
6431 source: forge_primitives::browser::HandOffSource::Feedback,
6432 client,
6433 };
6434
6435 let Command::RespondBrowserHandOff { answer: stamped, .. } = stamp(
6436 Command::RespondBrowserHandOff { key: seat.clone(), id, answer: answer(None) },
6437 7,
6438 ) else {
6439 panic!("the command is a hand-off answer");
6440 };
6441 assert_eq!(stamped.client, Some(7), "the connection that carried it is named");
6442
6443 let Command::RespondBrowserHandOff { answer: claimed, .. } =
6444 stamp(Command::RespondBrowserHandOff { key: seat, id, answer: answer(Some(99)) }, 7)
6445 else {
6446 panic!("the command is a hand-off answer");
6447 };
6448 assert_eq!(claimed.client, Some(7), "and a client cannot claim another's");
6449 }
6450
6451 #[test]
6456 fn only_its_own_takes_updates_are_heard() {
6457 let seat = SessionSlot::lead("TestOrg", "proj");
6458 let mine = SessionUpdate::DictateStarted {
6459 key: seat.clone(),
6460 floor_db: -50.0,
6461 generation: 1,
6462 initiator: Some(7),
6463 };
6464 assert!(ours_to_hear(&mine, 7), "a connection hears the take it started");
6465 assert!(!ours_to_hear(&mine, 8), "and not one another connection started");
6466
6467 let terminal = SessionUpdate::DictateLevel {
6468 key: seat,
6469 peak_db: -20.0,
6470 generation: 1,
6471 initiator: None,
6472 };
6473 assert!(
6474 !ours_to_hear(&terminal, 7),
6475 "the terminal's in-process take is nobody's on the socket"
6476 );
6477
6478 assert!(
6479 ours_to_hear(&SessionUpdate::CatalogLoaded, 7),
6480 "every other update keeps the watch rule alone"
6481 );
6482 }
6483
6484 #[test]
6491 fn a_new_claim_resets_before_the_older_take_ends() {
6492 let seat = SessionSlot::lead("TestOrg", "proj");
6493 let mut streaming = Streaming::default();
6494 streaming.claim(&seat);
6495 streaming.stamp(&seat, 1);
6496
6497 streaming.claim(&seat);
6500 streaming.retire(&seat, &DictateOutcome::Cancelled, 1);
6501 let claimed = streaming.seats.len();
6502 assert_eq!(claimed, 1, "the seat the live take streams into was stripped");
6503
6504 streaming.stamp(&seat, 2);
6507 streaming.retire(&seat, &DictateOutcome::Cancelled, 2);
6508 assert!(streaming.seats.is_empty(), "take 2's own end must retire the claim");
6509 }
6510
6511 #[test]
6520 fn a_stale_end_leaves_a_newer_claim() {
6521 let seat = SessionSlot::lead("TestOrg", "proj");
6522 let mut streaming = Streaming::default();
6523 streaming.claim(&seat);
6524 streaming.stamp(&seat, 1);
6525 streaming.claim(&seat);
6528 streaming.stamp(&seat, 2);
6529
6530 streaming.retire(&seat, &DictateOutcome::Cancelled, 1);
6531 assert_eq!(
6532 streaming.seats.len(),
6533 1,
6534 "the older take's end stripped the claim the newer take holds",
6535 );
6536 assert_eq!(streaming.seats[0].generation, 2, "and the stamp is the newer take's");
6537
6538 streaming.retire(&seat, &DictateOutcome::Cancelled, 2);
6540 assert!(streaming.seats.is_empty(), "the claim's own end must retire it");
6541 }
6542
6543 #[test]
6546 fn a_refusal_never_retires_a_claim() {
6547 let seat = SessionSlot::lead("TestOrg", "proj");
6548 let mut streaming = Streaming::default();
6549 streaming.claim(&seat);
6550
6551 streaming.retire(&seat, &DictateOutcome::Refused { message: "busy".to_owned() }, 0);
6554 assert_eq!(streaming.seats.len(), 1, "a refused start's end moved the claim");
6555
6556 streaming.stamp(&seat, 1);
6559 streaming.retire(&seat, &DictateOutcome::Refused { message: "busy".to_owned() }, 0);
6560 assert_eq!(streaming.seats.len(), 1, "a refusal's end moved the live claim");
6561 assert_eq!(streaming.seats[0].generation, 1, "and left its stamp alone");
6562 }
6563
6564 fn payload(samples: &[i16]) -> Vec<u8> {
6567 let mut bytes = vec![super::super::frame::Kind::Dictation.tag()];
6568 for sample in samples {
6569 bytes.extend_from_slice(&sample.to_le_bytes());
6570 }
6571 bytes
6572 }
6573
6574 #[test]
6578 fn a_frame_is_the_samples_its_own_connection_sent() {
6579 let seat = SessionSlot::lead("TestOrg", "proj");
6580 let streaming = Streaming { seats: vec![Claim { seat, generation: 0 }], read_aloud: false };
6581 assert_eq!(
6582 frame_route(&payload(&[16384]), &streaming),
6583 FrameRoute::Audio(vec![0.5]),
6584 "the samples the bytes carry"
6585 );
6586 }
6587
6588 #[test]
6591 fn a_frame_on_a_connection_with_no_take_goes_nowhere() {
6592 assert_eq!(frame_route(&payload(&[0]), &Streaming::default()), FrameRoute::NoTake);
6593 }
6594
6595 #[test]
6598 fn a_frame_lands_in_the_read_aloud_recording_with_no_seat() {
6599 let recording = Streaming { seats: Vec::new(), read_aloud: true };
6600
6601 assert_eq!(
6602 frame_route(&payload(&[16384]), &recording),
6603 FrameRoute::Audio(vec![0.5]),
6604 "a recording is a destination for a frame with no seat behind it"
6605 );
6606 assert_eq!(
6607 frame_route(&payload(&[16384]), &Streaming::default()),
6608 FrameRoute::NoTake,
6609 "a connection that is not recording has no destination for one"
6610 );
6611 }
6612
6613 #[test]
6616 fn a_binary_message_that_is_not_a_frame_is_refused_by_its_reason() {
6617 let seat = SessionSlot::lead("TestOrg", "proj");
6618 let streaming = Streaming { seats: vec![Claim { seat, generation: 0 }], read_aloud: false };
6619 assert_eq!(
6620 frame_route(&[], &streaming),
6621 FrameRoute::Refused(super::super::frame::Refusal::ShortHeader)
6622 );
6623 assert_eq!(
6624 frame_route(&[9, 0], &streaming),
6625 FrameRoute::Refused(super::super::frame::Refusal::UnknownKind(9))
6626 );
6627 }
6628
6629 #[test]
6634 fn a_document_answer_carries_the_bytes_and_the_kind_or_the_refusal() {
6635 use forge_primitives::document::{DocumentKind, DocumentRead};
6636 use forge_primitives::review::{
6637 ReviewAnchor, ReviewAuthor, ReviewComment, ReviewSide, ReviewStatus, ReviewThread,
6638 };
6639
6640 let (workspace, _dir) = crate::surface::testing::workspace();
6641 let surface = crate::surface::ViewSurface::new(Arc::clone(&workspace));
6642 let seat = SessionSlot::lead("TestOrg", "forge");
6643 let scope = forge_primitives::review::doc_scope_ref("/proj/docs/readme.md");
6647 let other = forge_primitives::review::doc_scope_ref("/proj/docs/other.md");
6648 let doc_thread = |id: &str, base_ref: &str| ReviewThread {
6649 id: id.to_owned(),
6650 anchor: ReviewAnchor {
6651 path: "/proj/docs/readme.md".to_owned(),
6652 side: ReviewSide::New,
6653 line: 1,
6654 content_hash: 7,
6655 context: vec!["# hi".to_owned()],
6656 base_ref: base_ref.to_owned(),
6657 },
6658 comments: vec![ReviewComment {
6659 author: ReviewAuthor::User,
6660 text: "one line, please".to_owned(),
6661 at: String::new(),
6662 review_id: None,
6663 }],
6664 status: ReviewStatus::Open,
6665 created_at: String::new(),
6666 updated_at: String::new(),
6667 commit: None,
6668 };
6669 assert!(workspace.upsert_review_thread("forge", &scope, doc_thread("d1", &scope)));
6670 assert!(workspace.upsert_review_thread("forge", &other, doc_thread("d2", &other)));
6671 workspace
6672 .submit_review(
6673 "forge",
6674 &scope,
6675 None,
6676 &["d1".to_owned()],
6677 SessionSlot::from_str_for_test("reader"),
6678 )
6679 .expect("a set for the document's scope");
6680
6681 let landed = Ok(DocumentRead {
6682 path: std::path::PathBuf::from("/proj/docs/readme.md"),
6683 kind: DocumentKind::Markdown,
6684 bytes: b"# hi\n".to_vec(),
6685 reviewable: true,
6686 content_hash: 42,
6687 });
6688
6689 let ServerMessage::DocumentRead {
6690 session,
6691 path,
6692 document_kind,
6693 bytes,
6694 reviewable,
6695 content_hash,
6696 threads,
6697 reviews,
6698 } = document_answer(&surface, &seat, landed)
6699 else {
6700 panic!("a read that landed is its own answer")
6701 };
6702 assert_eq!(session, seat, "the answer names the seat it was read for");
6703 assert_eq!(path, "/proj/docs/readme.md", "and the resolved absolute path");
6704 assert_eq!(document_kind, DocumentKind::Markdown, "and the kind the read detected");
6705 assert_eq!(bytes, "IyBoaQo=", "the bytes base64: what a client decodes is the document");
6706 assert!(reviewable, "and whether the document is reviewable");
6707 assert_eq!(content_hash, "42", "and the document's hash, as decimal digits");
6708 assert_eq!(threads.len(), 1, "the document's OWN comments ride the answer");
6709 assert_eq!(threads[0].thread.id, "d1");
6710 assert_eq!(threads[0].thread.anchor.base_ref, scope, "read under the document's scope");
6711 assert!(
6712 matches!(
6713 threads[0].resolution,
6714 Some(crate::transport::wire::ThreadResolutionWire::Placed {
6715 line: 1,
6716 moved_from: None,
6717 }),
6718 ),
6719 "and its captured line still sits where it was written: {:?}",
6720 threads[0].resolution,
6721 );
6722 assert_eq!(reviews.len(), 1, "and so do its review sets");
6723
6724 let revised = Ok(DocumentRead {
6728 path: std::path::PathBuf::from("/proj/docs/readme.md"),
6729 kind: DocumentKind::Markdown,
6730 bytes: b"# changed\n".to_vec(),
6731 reviewable: true,
6732 content_hash: 43,
6733 });
6734 let ServerMessage::DocumentRead { threads, .. } = document_answer(&surface, &seat, revised)
6735 else {
6736 panic!("a read that landed is its own answer")
6737 };
6738 assert_eq!(threads[0].thread.id, "d1", "the thread is still carried");
6739 assert!(
6740 matches!(
6741 threads[0].resolution,
6742 Some(crate::transport::wire::ThreadResolutionWire::Outdated {
6743 reason: crate::transport::wire::OutdatedReasonWire::Gone,
6744 }),
6745 ),
6746 "and the revision is reported, not guessed: {:?}",
6747 threads[0].resolution,
6748 );
6749
6750 let bommed = Ok(DocumentRead {
6756 path: std::path::PathBuf::from("/proj/docs/readme.md"),
6757 kind: DocumentKind::Markdown,
6758 bytes: b"\xef\xbb\xbf# hi\n".to_vec(),
6759 reviewable: true,
6760 content_hash: 44,
6761 });
6762 let ServerMessage::DocumentRead { threads, .. } = document_answer(&surface, &seat, bommed)
6763 else {
6764 panic!("a read that landed is its own answer")
6765 };
6766 assert!(
6767 matches!(
6768 threads[0].resolution,
6769 Some(crate::transport::wire::ThreadResolutionWire::Placed {
6770 line: 1,
6771 moved_from: None,
6772 }),
6773 ),
6774 "a BOM does not move the first line the client captured: {:?}",
6775 threads[0].resolution,
6776 );
6777
6778 let pdf_path = "/proj/docs/scan.pdf";
6784 let pdf_scope = forge_primitives::review::doc_scope_ref(pdf_path);
6785 let pdf_thread = ReviewThread {
6786 id: "p1".to_owned(),
6787 anchor: ReviewAnchor {
6788 path: pdf_path.to_owned(),
6789 side: ReviewSide::New,
6790 line: 1,
6791 content_hash: 9,
6792 context: vec!["a page".to_owned()],
6793 base_ref: pdf_scope.clone(),
6794 },
6795 comments: Vec::new(),
6796 status: ReviewStatus::Open,
6797 created_at: String::new(),
6798 updated_at: String::new(),
6799 commit: None,
6800 };
6801 assert!(workspace.upsert_review_thread("forge", &pdf_scope, pdf_thread));
6802 let pdf = Ok(DocumentRead {
6803 path: std::path::PathBuf::from(pdf_path),
6804 kind: DocumentKind::Pdf,
6805 bytes: b"%PDF-1.4\n\xff\xfe".to_vec(),
6806 reviewable: false,
6807 content_hash: 45,
6808 });
6809 let ServerMessage::DocumentRead { threads, .. } = document_answer(&surface, &seat, pdf)
6810 else {
6811 panic!("a read that landed is its own answer")
6812 };
6813 assert_eq!(threads.len(), 1, "the PDF scope's thread rides the answer");
6814 assert!(
6815 threads[0].resolution.is_none(),
6816 "a non-text read has no lines, so it carries no landing: {:?}",
6817 threads[0].resolution,
6818 );
6819
6820 let clean = Ok(DocumentRead {
6823 path: std::path::PathBuf::from("/proj/docs/third.md"),
6824 kind: DocumentKind::Text,
6825 bytes: b"x".to_vec(),
6826 reviewable: false,
6827 content_hash: 8,
6828 });
6829 let ServerMessage::DocumentRead { threads, reviews, .. } =
6830 document_answer(&surface, &seat, clean)
6831 else {
6832 panic!("a read that landed is its own answer")
6833 };
6834 assert!(threads.is_empty(), "no other document's comments leak in");
6835 assert!(reviews.is_empty(), "and no review sets either");
6836
6837 let refused = document_answer(
6838 &surface,
6839 &seat,
6840 Err("a PDF opens read-only; it cannot be handed reviewable".to_owned()),
6841 );
6842 let ServerMessage::Error { what, why, seat: named, .. } = refused else {
6843 panic!("a refused read is an error, not an empty document")
6844 };
6845 assert_eq!(what, "document_read", "the refusal names what was asked for");
6846 assert!(why.contains("read-only"), "and carries the read's own reason: {why}");
6847 assert_eq!(named, Some(seat), "under the seat that asked");
6848 }
6849
6850 #[test]
6857 fn a_refused_dispatch_echoes_the_dispatch_its_press_made() {
6858 let refused = dispatch_refusal(&DispatchError::PluginActionRunning, Some(41));
6859 let ServerMessage::Error { what, why, seat, dispatch_id } = refused else {
6860 panic!("a refusal is an error frame")
6861 };
6862 assert_eq!(what, "dispatch", "the tag a client routes a refusal by");
6863 assert!(
6864 why.contains("already running"),
6865 "in the core's own words, not this layer's: {why}"
6866 );
6867 assert_eq!(seat, None, "a plugin refusal belongs to no seat");
6868 assert_eq!(dispatch_id, Some(41), "and it names the dispatch the press made");
6869
6870 let refused = dispatch_refusal(&DispatchError::PluginActionRunning, None);
6871 let ServerMessage::Error { dispatch_id, .. } = refused else {
6872 panic!("a refusal is an error frame")
6873 };
6874 assert_eq!(dispatch_id, None, "a press that named none is echoed with none");
6875 }
6876
6877 fn transcript_rows(turns: usize) -> Vec<String> {
6881 (0..turns)
6882 .map(|at| {
6883 format!(
6884 "{{\"type\":\"user\",\"uuid\":\"u{at}\",\"session_id\":\"s1\",\
6885 \"message\":{{\"role\":\"user\",\"content\":\"turn {at}\"}}}}"
6886 )
6887 })
6888 .collect()
6889 }
6890
6891 fn transcript_frames(turns: usize) -> Vec<forge_primitives::Message> {
6893 (0..turns).map(turn_row).collect()
6894 }
6895
6896 fn live_frames(turns: usize) -> Vec<forge_primitives::Message> {
6900 let mut frames = Vec::with_capacity(turns * 2);
6901 for at in 0..turns {
6902 frames.push(turn_row(at));
6903 frames.push(result_frame(at));
6904 }
6905 frames
6906 }
6907
6908 fn turn_row(at: usize) -> forge_primitives::Message {
6910 serde_json::from_value(serde_json::json!({
6911 "type": "user",
6912 "uuid": format!("u{at}"),
6913 "message": {"role": "user", "content": format!("turn {at}")},
6914 "session_id": "s1",
6915 }))
6916 .expect("a user frame")
6917 }
6918
6919 fn calling_row(at: usize) -> forge_primitives::Message {
6921 serde_json::from_value(serde_json::json!({
6922 "type": "assistant",
6923 "uuid": format!("call{at}"),
6924 "message": {
6925 "id": format!("m{at}"),
6926 "role": "assistant",
6927 "model": "claude-opus-5",
6928 "content": [{
6929 "type": "tool_use",
6930 "id": format!("tu{at}"),
6931 "name": "Bash",
6932 "input": {"command": "ls"},
6933 }],
6934 },
6935 "session_id": "s1",
6936 }))
6937 .expect("an assistant frame")
6938 }
6939
6940 fn resulting_row(at: usize) -> forge_primitives::Message {
6942 serde_json::from_value(serde_json::json!({
6943 "type": "user",
6944 "uuid": format!("result{at}"),
6945 "message": {
6946 "role": "user",
6947 "content": [{
6948 "type": "tool_result",
6949 "tool_use_id": format!("tu{at}"),
6950 "content": "ok",
6951 }],
6952 },
6953 "session_id": "s1",
6954 }))
6955 .expect("a user frame")
6956 }
6957
6958 fn result_frame(at: usize) -> forge_primitives::Message {
6961 serde_json::from_value(serde_json::json!({
6962 "type": "result",
6963 "subtype": "success",
6964 "uuid": format!("r{at}"),
6965 "session_id": "s1",
6966 "is_error": false,
6967 "num_turns": at + 1,
6968 "duration_ms": 12,
6969 "duration_api_ms": 9,
6970 }))
6971 .expect("a result frame")
6972 }
6973
6974 fn turn_texts(page: &crate::transport::wire::Page) -> Vec<String> {
6976 page.turns
6977 .iter()
6978 .map(|turn| {
6979 turn.messages
6980 .iter()
6981 .find_map(|frame| {
6982 frame["message"]["content"][0]["text"].as_str().map(str::to_owned)
6983 })
6984 .unwrap_or_default()
6985 })
6986 .collect()
6987 }
6988
6989 #[test]
6995 fn a_transcript_page_hands_the_row_below_it_back_as_its_cursor() {
6996 let transcript = transcript_frames(160);
6999 let from = 148_usize;
7000 let span = forge_primitives::TranscriptSpan {
7001 messages: transcript[from..].to_vec(),
7002 offsets: (from as u64..160).map(|at| at * 100).collect(),
7003 exhausted: false,
7004 };
7005 let page = page_of_span(&span, 4, false, usize::MAX, None);
7006
7007 assert_eq!(
7008 turn_texts(&page),
7009 vec!["turn 156", "turn 157", "turn 158", "turn 159"],
7010 "the newest turns of the span",
7011 );
7012 assert_eq!(
7013 page.cursor.as_deref(),
7014 Some("r:15600:u156"),
7015 "and a cursor naming its oldest row: the byte it starts at, and its id",
7016 );
7017
7018 let short = forge_primitives::TranscriptSpan {
7023 messages: transcript[100..106].to_vec(),
7024 offsets: (100_u64..106).map(|at| at * 100).collect(),
7025 exhausted: false,
7026 };
7027 let short_page = page_of_span(&short, 8, false, usize::MAX, None);
7028 assert_eq!(short_page.turns.len(), 6, "precondition: the span holds six turns");
7029 assert_eq!(
7030 short_page.cursor.as_deref(),
7031 Some("r:10000:u100"),
7032 "a short span's page carries on from the row the span began with",
7033 );
7034 let ended = forge_primitives::TranscriptSpan { exhausted: true, ..short };
7035 assert!(
7036 page_of_span(&ended, 8, false, usize::MAX, None).cursor.is_none(),
7037 "and only a page reaching the transcript's first row ends the walk",
7038 );
7039 }
7040
7041 #[test]
7045 fn a_transcript_page_is_held_to_the_byte_budget() {
7046 let transcript = transcript_frames(8);
7047 let span = forge_primitives::TranscriptSpan {
7048 messages: transcript,
7049 offsets: (0u64..8).map(|at| at * 100).collect(),
7050 exhausted: true,
7051 };
7052 let unbounded = page_of_span(&span, 8, false, usize::MAX, None);
7053 assert_eq!(unbounded.turns.len(), 8, "precondition: the span holds eight turns");
7054
7055 let bounded = page_of_span(&span, 8, false, 1, None);
7056 assert!(
7057 !bounded.turns.is_empty() && bounded.turns.len() < 8,
7058 "a 1-byte budget keeps some of eight turns and not all: {}",
7059 bounded.turns.len(),
7060 );
7061 assert_eq!(
7062 turn_texts(&bounded).last().map(String::as_str),
7063 Some("turn 7"),
7064 "the newest turn is served even when it alone is over the budget",
7065 );
7066 assert!(
7067 bounded.cursor.is_some(),
7068 "and the row it stopped at is what the walk above continues from",
7069 );
7070 }
7071
7072 #[tokio::test]
7076 async fn a_page_below_the_floor_comes_from_the_transcript() {
7077 let dir = tempfile::tempdir().expect("a config dir of its own");
7078 let fleet =
7079 crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])]).expect("a fleet");
7080 fleet.start("TestOrg", "proj").expect("a live lead session");
7081 let rows = transcript_rows(80);
7082 fleet
7083 .seed_transcript(
7084 "TestOrg",
7085 "proj",
7086 "lead",
7087 &rows.iter().map(String::as_str).collect::<Vec<_>>(),
7088 )
7089 .expect("the session's transcript");
7090 let state = TransportState {
7091 surface: fleet.surface(),
7092 work: Arc::new(WorkCache::new()),
7093 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
7094 live: Mutex::new(crate::live::Live::new()),
7095 settings: crate::transport::settings_channel(),
7096 browser_host: None,
7097 heartbeat: Heartbeat::default(),
7098 };
7099 let seat = SessionSlot::lead("TestOrg", "proj");
7100 let cwd = state.surface.roster().cwd_for(&seat).expect("the seat's own directory");
7101
7102 let below = forge_primitives::TranscriptAnchor { row: "u40".to_owned(), offset: None };
7105 let BelowFloor::Page(page) = rows_below_floor(&state, &seat, &cwd, Some(below), 4).await
7106 else {
7107 panic!("the transcript answers a page below the floor")
7108 };
7109 assert_eq!(
7110 turn_texts(&page),
7111 vec!["turn 36", "turn 37", "turn 38", "turn 39"],
7112 "the newest turns below the row the walk stopped at",
7113 );
7114
7115 let cursor = page.cursor.clone().expect("a cursor the client asks below with");
7118 let anchor = row_cursor(&cursor).expect("a row cursor, which is what a page below mints");
7119 assert_eq!(anchor.row, "u36", "naming the page's own oldest row");
7120 assert!(anchor.offset.is_some(), "with the byte that row starts at");
7121 let BelowFloor::Page(below_page) =
7122 rows_below_floor(&state, &seat, &cwd, Some(anchor), 4).await
7123 else {
7124 panic!("the walk answers the page below the one it served")
7125 };
7126 assert_eq!(
7127 turn_texts(&below_page),
7128 vec!["turn 32", "turn 33", "turn 34", "turn 35"],
7129 "the turns directly above the page before it",
7130 );
7131
7132 let stranger =
7135 forge_primitives::TranscriptAnchor { row: "nowhere".to_owned(), offset: None };
7136 assert!(
7137 matches!(
7138 rows_below_floor(&state, &seat, &cwd, Some(stranger), 4).await,
7139 BelowFloor::Empty
7140 ),
7141 "an anchor the file does not carry is the empty page",
7142 );
7143
7144 let key = forge_workspace::userdata::catalog::scan::project_key_for_directory(cwd.to_str());
7146 let transcript = std::fs::read_dir(dir.path().join("projects").join(key))
7147 .expect("the project's transcript dir")
7148 .flatten()
7149 .map(|entry| entry.path())
7150 .find(|path| path.extension().is_some_and(|ext| ext == "jsonl"))
7151 .expect("the session's transcript");
7152 std::fs::remove_file(transcript).expect("the file goes");
7153 assert!(
7154 matches!(rows_below_floor(&state, &seat, &cwd, None, 4).await, BelowFloor::Empty),
7155 "a session whose file is gone is the empty page too",
7156 );
7157 }
7158
7159 #[tokio::test]
7163 async fn a_page_below_the_floor_is_held_to_the_more_page_budget() {
7164 let dir = tempfile::tempdir().expect("a config dir of its own");
7165 let fleet = crate::testing::Fleet::in_dir_with_server(
7166 dir.path(),
7167 &[("TestOrg", &["proj"])],
7168 "[server]\nmore_page_cap_kb = 1\n",
7169 )
7170 .expect("a fleet");
7171 fleet.start("TestOrg", "proj").expect("a live lead session");
7172 let rows: Vec<String> = (0..12)
7173 .map(|at| {
7174 format!(
7175 "{{\"type\":\"user\",\"uuid\":\"u{at}\",\"session_id\":\"s1\",\
7176 \"message\":{{\"role\":\"user\",\"content\":\"turn {at} {}\"}}}}",
7177 "x".repeat(2_000),
7178 )
7179 })
7180 .collect();
7181 fleet
7182 .seed_transcript(
7183 "TestOrg",
7184 "proj",
7185 "lead",
7186 &rows.iter().map(String::as_str).collect::<Vec<_>>(),
7187 )
7188 .expect("the session's transcript");
7189 let state = TransportState {
7190 surface: fleet.surface(),
7191 work: Arc::new(WorkCache::new()),
7192 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
7193 live: Mutex::new(crate::live::Live::new()),
7194 settings: crate::transport::settings_channel(),
7195 browser_host: None,
7196 heartbeat: Heartbeat::default(),
7197 };
7198 let seat = SessionSlot::lead("TestOrg", "proj");
7199 let cwd = state.surface.roster().cwd_for(&seat).expect("the seat's own directory");
7200 let below = forge_primitives::TranscriptAnchor { row: "u8".to_owned(), offset: None };
7201 let BelowFloor::Page(page) = rows_below_floor(&state, &seat, &cwd, Some(below), 8).await
7202 else {
7203 panic!("the transcript answers a page below the floor")
7204 };
7205 let texts = turn_texts(&page);
7206 assert_eq!(
7207 texts.len(),
7208 1,
7209 "a 1 KB budget keeps the newest turn of eight large ones and not the rest: {texts:?}",
7210 );
7211 assert!(
7212 texts[0].starts_with("turn 7"),
7213 "and it is the newest row below the anchor: {:?}",
7214 texts[0],
7215 );
7216 assert!(
7217 page.cursor.is_some(),
7218 "and the walk above it continues from the row it stopped at",
7219 );
7220 }
7221
7222 #[tokio::test]
7227 async fn a_zero_more_page_cap_caps_the_records_below_the_floor() {
7228 let dir = tempfile::tempdir().expect("a config dir of its own");
7229 let fleet = crate::testing::Fleet::in_dir_with_server(
7230 dir.path(),
7231 &[("TestOrg", &["proj"])],
7232 "[server]\nmore_page_cap_kb = 0\ntool_record_cap_kb = 1\n",
7233 )
7234 .expect("a fleet");
7235 fleet.start("TestOrg", "proj").expect("a live lead session");
7236 let mut rows = transcript_rows(80);
7237 let big = [
7240 "{\"type\":\"user\",\"uuid\":\"u23\",\"session_id\":\"s1\",\
7241 \"message\":{\"role\":\"user\",\"content\":\"turn 23\"}}"
7242 .to_owned(),
7243 "{\"type\":\"assistant\",\"uuid\":\"a23\",\
7244 \"message\":{\"id\":\"m23\",\"role\":\"assistant\",\
7245 \"model\":\"claude-opus-5\",\
7246 \"content\":[{\"type\":\"text\",\"text\":\"reply 23\"}]}}"
7247 .to_owned(),
7248 format!(
7249 "{{\"type\":\"user\",\"uuid\":\"t23\",\"session_id\":\"s1\",\
7250 \"message\":{{\"role\":\"user\",\"content\":[{{\"type\":\"tool_result\",\
7251 \"tool_use_id\":\"call-23\",\"content\":\"ok\"}}]}},\
7252 \"toolUseResult\":{{\"payload\":\"{}\"}}}}",
7253 "x".repeat(20_000),
7254 ),
7255 "{\"type\":\"result\",\"uuid\":\"r23\",\"subtype\":\"success\",\
7256 \"duration_ms\":1,\"duration_api_ms\":1,\"is_error\":false,\
7257 \"num_turns\":1,\"session_id\":\"s1\"}"
7258 .to_owned(),
7259 ];
7260 rows.splice(23..24, big);
7261 fleet
7262 .seed_transcript(
7263 "TestOrg",
7264 "proj",
7265 "lead",
7266 &rows.iter().map(String::as_str).collect::<Vec<_>>(),
7267 )
7268 .expect("the session's transcript");
7269 let state = TransportState {
7270 surface: fleet.surface(),
7271 work: Arc::new(WorkCache::new()),
7272 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
7273 live: Mutex::new(crate::live::Live::new()),
7274 settings: crate::transport::settings_channel(),
7275 browser_host: None,
7276 heartbeat: Heartbeat::default(),
7277 };
7278 let seat = SessionSlot::lead("TestOrg", "proj");
7279 let cwd = state.surface.roster().cwd_for(&seat).expect("the seat's own directory");
7280 let below = forge_primitives::TranscriptAnchor { row: "u25".to_owned(), offset: None };
7281 let BelowFloor::Page(page) = rows_below_floor(&state, &seat, &cwd, Some(below), 4).await
7282 else {
7283 panic!("the transcript answers a page below the floor")
7284 };
7285 assert_eq!(
7286 turn_texts(&page),
7287 vec!["turn 21", "turn 22", "turn 23", "turn 24"],
7288 "the newest turns below the anchor",
7289 );
7290 let record = page
7291 .turns
7292 .iter()
7293 .flat_map(|turn| turn.messages.iter())
7294 .find(|frame| frame["uuid"] == "t23")
7295 .expect("the record's frame is in the page");
7296 assert_eq!(
7297 record["tool_use_result"]["omitted"], true,
7298 "the record is capped below the floor: {record}",
7299 );
7300 }
7301
7302 #[tokio::test]
7310 async fn a_page_below_a_rowless_window_serves_the_rows_beneath_it() {
7311 const TURNS: usize = 2_600;
7315 let dir = tempfile::tempdir().expect("a config dir of its own");
7316 let fleet =
7317 crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])]).expect("a fleet");
7318 fleet.start("TestOrg", "proj").expect("a live lead session");
7319 let rows = transcript_rows(TURNS);
7320 fleet
7321 .seed_transcript(
7322 "TestOrg",
7323 "proj",
7324 "lead",
7325 &rows.iter().map(String::as_str).collect::<Vec<_>>(),
7326 )
7327 .expect("the session's transcript");
7328 let held = crate::transport::conversation::Held::new(
7329 crate::transport::conversation::Conversation::new(live_frames(TURNS), 0, 0),
7330 );
7331 assert!(
7332 held.lock().dropped() > 0,
7333 "precondition: the live conversation outgrew its window"
7334 );
7335 assert!(
7336 held.lock()
7337 .messages()
7338 .iter()
7339 .any(|message| !forge_workspace::has_a_transcript_row(message)),
7340 "precondition: the conversation carries frames no transcript row has",
7341 );
7342 let state = TransportState {
7343 surface: fleet.surface(),
7344 work: Arc::new(WorkCache::new()),
7345 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
7346 live: Mutex::new(crate::live::Live::new()),
7347 settings: crate::transport::settings_channel(),
7348 browser_host: None,
7349 heartbeat: Heartbeat::default(),
7350 };
7351 let seat = SessionSlot::lead("TestOrg", "proj");
7352 let cwd = state.surface.roster().cwd_for(&seat).expect("the seat's own directory");
7353
7354 let boundary = boundary_of(&held.lock()).expect("the window carries rows");
7355 let oldest =
7356 boundary.row.trim_start_matches('u').parse::<usize>().expect("a row id names its turn");
7357 assert!(boundary.offset.is_none(), "named by id alone, which is what the search is for");
7358 let BelowFloor::Page(page) = rows_below_floor(&state, &seat, &cwd, Some(boundary), 4).await
7359 else {
7360 panic!("the transcript answers a page below the window's oldest row")
7361 };
7362 assert_eq!(
7363 turn_texts(&page),
7364 (oldest - 4..oldest).map(|at| format!("turn {at}")).collect::<Vec<_>>(),
7365 "the rows directly beneath the window's oldest row, with no row stepped over for the \
7366 result frames in between",
7367 );
7368 }
7369
7370 #[test]
7377 fn a_stop_never_lands_on_a_result_without_its_call() {
7378 let mut frames: Vec<forge_primitives::Message> = Vec::new();
7381 for at in 0..4_001 {
7382 frames.push(calling_row(at));
7383 frames.push(resulting_row(at));
7384 }
7385 frames.push(calling_row(9_999));
7389 let held = crate::transport::conversation::Held::new(
7390 crate::transport::conversation::Conversation::new(frames, 0, 0),
7391 );
7392 let window = held.lock();
7393 assert!(window.dropped() > 0, "precondition: the conversation outgrew its window");
7394 assert!(
7395 carries_a_result(window.messages().first().expect("a window")),
7396 "precondition: the count cut landed on a result row",
7397 );
7398
7399 let boundary = boundary_of(&window).expect("the window carries rows");
7400
7401 assert_eq!(
7402 boundary.row, "call2002",
7403 "the stop stepped past the result the cut landed on, so the page below carries it \
7404 with the call that produced it",
7405 );
7406 }
7407
7408 #[tokio::test]
7414 async fn a_page_from_the_middle_of_a_large_transcript_seeks_by_its_cursor() {
7415 const ROWS: usize = 13_000;
7416 let dir = tempfile::tempdir().expect("a config dir of its own");
7417 let fleet =
7418 crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])]).expect("a fleet");
7419 fleet.start("TestOrg", "proj").expect("a live lead session");
7420 let rows = transcript_rows(ROWS);
7421 fleet
7422 .seed_transcript(
7423 "TestOrg",
7424 "proj",
7425 "lead",
7426 &rows.iter().map(String::as_str).collect::<Vec<_>>(),
7427 )
7428 .expect("the session's transcript");
7429 let held = crate::transport::conversation::Held::new(
7430 crate::transport::conversation::Conversation::new(transcript_frames(ROWS), 0, 0),
7431 );
7432 assert!(held.lock().dropped() > 0, "precondition: the copy outgrew its window");
7433 let state = TransportState {
7434 surface: fleet.surface(),
7435 work: Arc::new(WorkCache::new()),
7436 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
7437 live: Mutex::new(crate::live::Live::new()),
7438 settings: crate::transport::settings_channel(),
7439 browser_host: None,
7440 heartbeat: Heartbeat::default(),
7441 };
7442 let seat = SessionSlot::lead("TestOrg", "proj");
7443 let cwd = state.surface.roster().cwd_for(&seat).expect("the seat's own directory");
7444
7445 let boundary = boundary_of(&held.lock()).expect("the window carries rows");
7446 let BelowFloor::Page(page) = rows_below_floor(&state, &seat, &cwd, Some(boundary), 4).await
7447 else {
7448 panic!("a page below the floor is answered")
7449 };
7450 let cursor = page.cursor.clone().expect("a cursor the client asks below with");
7451 let anchor = row_cursor(&cursor).expect("a row cursor");
7452 assert!(anchor.offset.is_some(), "carrying the byte that row starts at, for the seek");
7453 let oldest =
7454 anchor.row.trim_start_matches('u').parse::<usize>().expect("a row id names its turn");
7455 assert_eq!(
7456 turn_texts(&page).first().map(String::as_str),
7457 Some(format!("turn {oldest}").as_str()),
7458 "the page begins at the row its cursor names",
7459 );
7460
7461 let BelowFloor::Page(below) = rows_below_floor(&state, &seat, &cwd, Some(anchor), 4).await
7462 else {
7463 panic!("the page below it is answered too")
7464 };
7465 assert_eq!(
7466 turn_texts(&below).last().map(String::as_str),
7467 Some(format!("turn {}", oldest - 1).as_str()),
7468 "and the page below ends on the turn directly beneath where the first began",
7469 );
7470 }
7471
7472 fn subagent_frame(at: usize) -> forge_primitives::Message {
7475 serde_json::from_value(serde_json::json!({
7476 "type": "assistant",
7477 "uuid": format!("sub{at}"),
7478 "parent_tool_use_id": "toolu_dispatch",
7479 "session_id": "s1",
7480 "message": {
7481 "id": format!("msub{at}"),
7482 "role": "assistant",
7483 "model": "claude-opus-5",
7484 "content": [{"type": "text", "text": format!("inside the sub-agent {at}")}],
7485 },
7486 }))
7487 .expect("an assistant frame")
7488 }
7489
7490 #[tokio::test]
7501 async fn a_window_front_the_file_cannot_name_still_pages() {
7502 let dir = tempfile::tempdir().expect("a config dir of its own");
7503 let fleet =
7504 crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])]).expect("a fleet");
7505 fleet.start("TestOrg", "proj").expect("a live lead session");
7506 let seat = SessionSlot::lead("TestOrg", "proj");
7507
7508 let mut frames = live_frames(2_600);
7514 frames.extend((0..999).map(result_frame));
7515 let conversations = crate::transport::conversation::Conversations::new();
7516 conversations
7517 .insert(&seat, crate::transport::conversation::Conversation::new(frames, 0, 0));
7518 conversations.apply(
7519 &SessionUpdate::CronPromptAppended {
7520 key: seat.clone(),
7521 text: "the cron fired".to_owned(),
7522 uuid: "cap-cron".to_owned(),
7523 cron_id: "c1".to_owned(),
7524 description: None,
7525 },
7526 None,
7527 );
7528 let following = resulting_row(9_999);
7529 conversations.apply(
7530 &SessionUpdate::ChatAppended {
7531 key: seat.clone(),
7532 msg: following.clone(),
7533 source: None,
7534 },
7535 None,
7536 );
7537 for at in 0..5_000_usize {
7538 let window = conversations.get(&seat).expect("the seat is held");
7539 if row_id(window.lock().messages().first().expect("a front")).as_deref()
7540 == Some("cap-cron")
7541 {
7542 break;
7543 }
7544 drop(window);
7545 conversations.apply(
7546 &SessionUpdate::ChatAppended {
7547 key: seat.clone(),
7548 msg: result_frame(at),
7549 source: None,
7550 },
7551 None,
7552 );
7553 }
7554 let held = conversations.get(&seat).expect("the seat is held");
7555 let boundary = {
7556 let window = held.lock();
7557 assert_eq!(
7558 row_id(window.messages().first().expect("a front")).as_deref(),
7559 Some("cap-cron"),
7560 "precondition: the count cut landed on the forged row",
7561 );
7562 boundary_of(&window).expect("the window carries a row the file has")
7563 };
7564 assert_eq!(
7565 boundary.row, "result9999",
7566 "the forged row is passed over for the first row under it the file carries",
7567 );
7568
7569 let rows = transcript_rows(2_600);
7572 let mut rows = rows;
7573 rows.push(serde_json::to_string(&following).expect("the following row as a line"));
7574 fleet
7575 .seed_transcript(
7576 "TestOrg",
7577 "proj",
7578 "lead",
7579 &rows.iter().map(String::as_str).collect::<Vec<_>>(),
7580 )
7581 .expect("the session's transcript");
7582 let state = TransportState {
7583 surface: fleet.surface(),
7584 work: Arc::new(WorkCache::new()),
7585 conversations: Arc::new(conversations),
7586 live: Mutex::new(crate::live::Live::new()),
7587 settings: crate::transport::settings_channel(),
7588 browser_host: None,
7589 heartbeat: Heartbeat::default(),
7590 };
7591 let cwd = state.surface.roster().cwd_for(&seat).expect("the seat's own directory");
7592 let BelowFloor::Page(page) = rows_below_floor(&state, &seat, &cwd, Some(boundary), 4).await
7593 else {
7594 panic!("the transcript answers a page below a front it was named a real row for")
7595 };
7596 assert!(!page.turns.is_empty(), "and the page is not the empty one");
7597
7598 let mut frames = live_frames(2_600);
7603 frames.extend((0..5_400).map(subagent_frame));
7604 let held = crate::transport::conversation::Held::new(
7605 crate::transport::conversation::Conversation::new(frames, 0, 0),
7606 );
7607 let window = held.lock();
7608 let front = window.messages().first().expect("a front");
7609 assert!(
7610 row_id(front).is_some_and(|id| id.starts_with("sub")),
7611 "precondition: the count cut landed inside the sub-agent run",
7612 );
7613 assert!(
7614 !forge_workspace::has_a_transcript_row(front),
7615 "precondition: which the file counts no row for",
7616 );
7617 assert!(
7618 boundary_of(&window).is_none(),
7619 "and the boundary passes over every such frame, naming nothing to the file: {:?}",
7620 boundary_of(&window),
7621 );
7622 }
7623
7624 fn unnameable_row() -> forge_primitives::Message {
7628 serde_json::from_value(serde_json::json!({
7629 "type": "user",
7630 "session_id": "s1",
7631 "message": {"role": "user", "content": "no id on this row"},
7632 }))
7633 .expect("a user frame")
7634 }
7635
7636 fn is_unnameable(message: &forge_primitives::Message) -> bool {
7638 serde_json::to_value(message)
7639 .ok()
7640 .and_then(|value| value["message"]["content"][0]["text"].as_str().map(str::to_owned))
7641 .as_deref()
7642 == Some("no id on this row")
7643 }
7644
7645 #[tokio::test]
7655 async fn a_front_the_file_cannot_be_named_by_still_pages_to_the_row_behind_it() {
7656 let dir = tempfile::tempdir().expect("a config dir of its own");
7657 let fleet =
7658 crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])]).expect("a fleet");
7659 fleet.start("TestOrg", "proj").expect("a live lead session");
7660 let seat = SessionSlot::lead("TestOrg", "proj");
7661
7662 let mut frames = live_frames(2_600);
7665 frames.extend((0..999).map(result_frame));
7666 let conversations = crate::transport::conversation::Conversations::new();
7667 conversations
7668 .insert(&seat, crate::transport::conversation::Conversation::new(frames, 0, 0));
7669 conversations.apply(
7670 &SessionUpdate::ChatAppended { key: seat.clone(), msg: unnameable_row(), source: None },
7671 None,
7672 );
7673 conversations.apply(
7674 &SessionUpdate::ChatAppended {
7675 key: seat.clone(),
7676 msg: resulting_row(9_999),
7677 source: None,
7678 },
7679 None,
7680 );
7681 for at in 0..5_000_usize {
7682 let window = conversations.get(&seat).expect("the seat is held");
7683 let front = window.lock().messages().first().cloned().expect("a front");
7684 if is_unnameable(&front) {
7685 break;
7686 }
7687 conversations.apply(
7688 &SessionUpdate::ChatAppended {
7689 key: seat.clone(),
7690 msg: result_frame(at),
7691 source: None,
7692 },
7693 None,
7694 );
7695 }
7696 let held = conversations.get(&seat).expect("the seat is held");
7697 let window = held.lock();
7698 let front = window.messages().first().expect("a front");
7699 assert!(
7700 is_unnameable(front),
7701 "precondition: the count cut landed on the row with no id ({:?})",
7702 row_id(front),
7703 );
7704 let boundary = boundary_of(&window);
7705 assert_eq!(
7706 boundary.as_ref().map(|anchor| anchor.row.as_str()),
7707 Some("result9999"),
7708 "the boundary names the first row behind the front that the file both carries and can \
7709 be asked for: {boundary:?}",
7710 );
7711 }
7712
7713 #[tokio::test]
7724 async fn a_walk_below_the_floor_reaches_the_transcripts_start() {
7725 const TURNS: usize = 2_600;
7726 let dir = tempfile::tempdir().expect("a config dir of its own");
7727 let fleet =
7728 crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])]).expect("a fleet");
7729 fleet.start("TestOrg", "proj").expect("a live lead session");
7730 let rows = transcript_rows(TURNS);
7731 fleet
7732 .seed_transcript(
7733 "TestOrg",
7734 "proj",
7735 "lead",
7736 &rows.iter().map(String::as_str).collect::<Vec<_>>(),
7737 )
7738 .expect("the session's transcript");
7739 let held = crate::transport::conversation::Held::new(
7740 crate::transport::conversation::Conversation::new(live_frames(TURNS), 0, 0),
7741 );
7742 assert!(
7743 held.lock()
7744 .messages()
7745 .iter()
7746 .any(|message| !forge_workspace::has_a_transcript_row(message)),
7747 "precondition: the walk crosses frames with no row",
7748 );
7749 let state = TransportState {
7750 surface: fleet.surface(),
7751 work: Arc::new(WorkCache::new()),
7752 conversations: Arc::new(crate::transport::conversation::Conversations::new()),
7753 live: Mutex::new(crate::live::Live::new()),
7754 settings: crate::transport::settings_channel(),
7755 browser_host: None,
7756 heartbeat: Heartbeat::default(),
7757 };
7758 let seat = SessionSlot::lead("TestOrg", "proj");
7759 let cwd = state.surface.roster().cwd_for(&seat).expect("the seat's own directory");
7760
7761 let mut below = Some(boundary_of(&held.lock()).expect("the window carries rows"));
7762 let mut pages = 0_usize;
7763 let mut served: Vec<Vec<usize>> = Vec::new();
7769 loop {
7770 let BelowFloor::Page(page) = rows_below_floor(&state, &seat, &cwd, below, 20).await
7771 else {
7772 panic!("the walk's page {pages} is answered from the transcript");
7773 };
7774 let texts = turn_texts(&page);
7775 assert!(!texts.is_empty(), "the walk's page {pages} holds turns");
7776 let numbers: Vec<usize> = texts
7779 .iter()
7780 .map(|text| text.trim_start_matches("turn ").parse().expect("a turn's own number"))
7781 .collect();
7782 assert!(
7783 numbers.windows(2).all(|pair| pair[0] < pair[1]),
7784 "the walk's page {pages} ascends: {numbers:?}",
7785 );
7786 if let Some(previous) = served.last() {
7790 let prev_oldest = *previous.first().unwrap_or(&0);
7792 let next_newest = *numbers.last().unwrap_or(&0);
7793 assert!(
7794 next_newest + 1 >= prev_oldest,
7795 "page {pages} leaves no hole below the page above: {previous:?} then \
7796 {numbers:?}",
7797 );
7798 assert!(
7799 next_newest < prev_oldest,
7800 "page {pages} sits below the page above: {previous:?} then {numbers:?}",
7801 );
7802 }
7803 pages += 1;
7804 assert!(pages < 200, "the walk reached the transcript's beginning");
7805 served.push(numbers);
7806 match page.cursor.as_deref().and_then(row_cursor) {
7807 Some(next) => below = Some(next),
7808 None => break,
7809 }
7810 }
7811 assert_eq!(
7812 served.last().and_then(|page| page.first()).copied(),
7813 Some(0),
7814 "and the last page it served runs down to the transcript's own first turn",
7815 );
7816 let messages = state.surface.conversation(&seat, &cwd).messages;
7820 let rendered = crate::transcript::render(&messages, false);
7821 let expected: Vec<usize> = turn_texts(&crate::transport::wire::Page {
7822 turns: crate::transport::wire::all_turns(&messages, &rendered),
7823 cursor: None,
7824 })
7825 .iter()
7826 .map(|text| text.trim_start_matches("turn ").parse().expect("a turn's own number"))
7827 .collect();
7828 let window = held.lock();
7829 let window_numbers: Vec<usize> = turn_texts(&crate::transport::wire::Page {
7830 turns: crate::transport::wire::all_turns(
7831 window.messages(),
7832 &crate::transcript::render(window.messages(), false),
7833 ),
7834 cursor: None,
7835 })
7836 .iter()
7837 .map(|text| text.trim_start_matches("turn ").parse().expect("a turn's own number"))
7838 .collect();
7839 drop(window);
7840 let served: std::collections::HashSet<usize> =
7841 served.into_iter().flatten().chain(window_numbers).collect();
7842 for turn in &expected {
7843 assert!(served.contains(turn), "the walk serves turn {turn}");
7844 }
7845 }
7846
7847 struct Peer {
7850 inbound: mpsc::UnboundedReceiver<Message>,
7851 outbound: mpsc::UnboundedSender<Message>,
7852 }
7853
7854 impl futures_util::Stream for Peer {
7855 type Item = Result<Message, axum::Error>;
7856 fn poll_next(
7857 mut self: std::pin::Pin<&mut Self>,
7858 cx: &mut std::task::Context<'_>,
7859 ) -> std::task::Poll<Option<Self::Item>> {
7860 self.inbound.poll_recv(cx).map(|item| item.map(Ok))
7861 }
7862 }
7863
7864 impl futures_util::Sink<Message> for Peer {
7865 type Error = axum::Error;
7866 fn poll_ready(
7867 self: std::pin::Pin<&mut Self>,
7868 _cx: &mut std::task::Context<'_>,
7869 ) -> std::task::Poll<Result<(), Self::Error>> {
7870 std::task::Poll::Ready(Ok(()))
7871 }
7872 fn start_send(self: std::pin::Pin<&mut Self>, item: Message) -> Result<(), Self::Error> {
7873 let _ = self.outbound.send(item);
7874 Ok(())
7875 }
7876 fn poll_flush(
7877 self: std::pin::Pin<&mut Self>,
7878 _cx: &mut std::task::Context<'_>,
7879 ) -> std::task::Poll<Result<(), Self::Error>> {
7880 std::task::Poll::Ready(Ok(()))
7881 }
7882 fn poll_close(
7883 self: std::pin::Pin<&mut Self>,
7884 _cx: &mut std::task::Context<'_>,
7885 ) -> std::task::Poll<Result<(), Self::Error>> {
7886 std::task::Poll::Ready(Ok(()))
7887 }
7888 }
7889
7890 struct Driven {
7893 to_server: mpsc::UnboundedSender<Message>,
7894 written: Arc<Mutex<Vec<ServerMessage>>>,
7895 driven: tokio::task::JoinHandle<anyhow::Result<()>>,
7896 progress: Arc<Progress>,
7899 }
7900
7901 fn connection(state: &Arc<TransportState>) -> Driven {
7902 let (to_server, inbound) = mpsc::unbounded_channel();
7903 let (outbound, mut from_server) = mpsc::unbounded_channel();
7904 let peer = ClientSocket::new(Peer { inbound, outbound }, false);
7905 let written = Arc::new(Mutex::new(Vec::new()));
7906 let seen = Arc::clone(&written);
7907 tokio::spawn(async move {
7910 while let Some(Message::Text(text)) = from_server.recv().await {
7911 if let Ok(message) = serde_json::from_str::<ServerMessage>(&text) {
7912 seen.lock().expect("the frame log").push(message);
7913 }
7914 }
7915 });
7916 let state = Arc::clone(state);
7917 let settings = crate::transport::settings_channel();
7918 let progress = Arc::new(Progress::new());
7921 let watched_progress = Arc::clone(&progress);
7922 let driven = tokio::spawn(async move {
7925 let receiver = settings.subscribe();
7926 let mut peer = peer.watched(Arc::clone(&watched_progress));
7927 supervise(&mut peer, &state, attached_from(), receiver, None, &watched_progress).await
7928 });
7929 Driven { to_server, written, driven, progress }
7930 }
7931
7932 async fn until(condition: impl Fn() -> bool, why: &str) {
7934 tokio::time::timeout(Duration::from_secs(5), async {
7935 while !condition() {
7936 tokio::task::yield_now().await;
7937 }
7938 })
7939 .await
7940 .unwrap_or_else(|_| panic!("timed out waiting for {why}"));
7941 }
7942
7943 fn home_snapshots(written: &Arc<Mutex<Vec<ServerMessage>>>) -> usize {
7944 written
7945 .lock()
7946 .expect("the frame log")
7947 .iter()
7948 .filter(|frame| matches!(frame, ServerMessage::Snapshot { subject: Subject::Home, .. }))
7949 .count()
7950 }
7951
7952 async fn subscribed_home(
7954 to_server: &mpsc::UnboundedSender<Message>,
7955 written: &Arc<Mutex<Vec<ServerMessage>>>,
7956 ) {
7957 to_server
7958 .send(Message::Text(
7959 serde_json::to_string(&ClientMessage::Subscribe {
7960 what: Subject::Home,
7961 answering: false,
7962 review: false,
7963 })
7964 .expect("the subscribe encodes")
7965 .into(),
7966 ))
7967 .expect("the message goes");
7968 until(|| home_snapshots(written) == 1, "the home's first read").await;
7969 }
7970
7971 async fn subscribed_home_again(
7975 to_server: &mpsc::UnboundedSender<Message>,
7976 written: &Arc<Mutex<Vec<ServerMessage>>>,
7977 ) -> serde_json::Value {
7978 let before = home_snapshots(written);
7979 to_server
7980 .send(Message::Text(
7981 serde_json::to_string(&ClientMessage::Subscribe {
7982 what: Subject::Home,
7983 answering: false,
7984 review: false,
7985 })
7986 .expect("the subscribe encodes")
7987 .into(),
7988 ))
7989 .expect("the message goes");
7990 until(|| home_snapshots(written) == before + 1, "the barrier home read").await;
7991 written
7994 .lock()
7995 .expect("the frame log")
7996 .iter()
7997 .rev()
7998 .find_map(|frame| match frame {
7999 ServerMessage::Snapshot { subject: Subject::Home, data } => Some(data.clone()),
8000 _ => None,
8001 })
8002 .expect("the barrier's home snapshot")
8003 }
8004
8005 fn seat_read(written: &Arc<Mutex<Vec<ServerMessage>>>) -> bool {
8007 written.lock().expect("the frame log").iter().any(|frame| {
8008 matches!(frame, ServerMessage::Snapshot { subject: Subject::Session(_), .. })
8009 })
8010 }
8011
8012 fn arm_diamond(state: &Arc<TransportState>, slot: &SessionSlot) {
8014 let result = serde_json::from_value(serde_json::json!({
8015 "type": "result",
8016 "subtype": "success",
8017 "duration_ms": 1,
8018 "duration_api_ms": 1,
8019 "is_error": false,
8020 "num_turns": 1,
8021 "session_id": "s",
8022 }))
8023 .expect("a result message");
8024 Live::lock(&state.live).apply(&SessionUpdate::ChatAppended {
8025 key: slot.clone(),
8026 msg: result,
8027 source: None,
8028 });
8029 }
8030
8031 fn showing(to_server: &mpsc::UnboundedSender<Message>, seat: Option<SessionSlot>) {
8032 to_server
8033 .send(Message::Text(
8034 serde_json::to_string(&ClientMessage::Showing { seat })
8035 .expect("the showing encodes")
8036 .into(),
8037 ))
8038 .expect("the message goes");
8039 }
8040
8041 fn subscribed(to_server: &mpsc::UnboundedSender<Message>, slot: &SessionSlot) {
8042 to_server
8043 .send(Message::Text(
8044 serde_json::to_string(&ClientMessage::Subscribe {
8045 what: Subject::Session(slot.clone()),
8046 answering: false,
8047 review: false,
8048 })
8049 .expect("the subscribe encodes")
8050 .into(),
8051 ))
8052 .expect("the message goes");
8053 }
8054
8055 fn lead() -> SessionSlot {
8056 SessionSlot::lead("TestOrg", "proj")
8057 }
8058
8059 #[tokio::test]
8065 async fn a_seat_read_carries_the_cron_pairing_its_rows_came_from() {
8066 let dir = tempfile::tempdir().expect("a config dir of its own");
8067 let app_support = dir.path().join("app-support");
8070 std::fs::create_dir_all(&app_support).expect("app-support");
8071 {
8072 let db_path = forge_workspace::store::path_for(&app_support, dir.path());
8074 std::fs::create_dir_all(db_path.parent().expect("stores/")).expect("stores/");
8075 let db = forge_workspace::store::Db::open(&db_path).expect("open db");
8076 forge_workspace::store::cron_names::put(
8077 &db,
8078 "p-cron",
8079 &forge_primitives::cron::CronName {
8080 cron_id: "c1".to_owned(),
8081 description: Some("Morning summary".to_owned()),
8082 },
8083 std::time::SystemTime::now(),
8084 )
8085 .expect("write the pairing");
8086 }
8087 let fleet = crate::testing::Fleet::in_dir(dir.path(), &[("TestOrg", &["proj"])])
8088 .expect("the fleet builds");
8089 let cron_row = r#"{"type":"user","uuid":"p-cron","message":{"role":"user","content":[{"type":"text","text":"[Cron]\n\nsummarise overnight CI"}]}}"#;
8093 let after_row = r#"{"type":"assistant","uuid":"a-after","message":{"id":"m-after","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"done"}]}}"#;
8094 fleet
8095 .seed_transcript("TestOrg", "proj", "lead", &[cron_row, after_row])
8096 .expect("the fired prompt's row and the one after it");
8097 let below_the_second_row = format!("r:{}:a-after", cron_row.len() + 1);
8099 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8100 fleet.hold_conversation(&state, "TestOrg", "proj", "lead").expect("the held conversation");
8101 let slot = lead();
8102 let Driven { to_server, written, driven, .. } = connection(&state);
8103
8104 subscribed(&to_server, &slot);
8105 until(
8106 || {
8107 written.lock().expect("the frame log").iter().any(|frame| {
8108 let ServerMessage::Snapshot { subject: Subject::Session(read), data } = frame
8109 else {
8110 return false;
8111 };
8112 read == &slot
8113 && data
8114 .pointer("/conversation/cron_names/p-cron/description")
8115 .and_then(serde_json::Value::as_str)
8116 == Some("Morning summary")
8117 })
8118 },
8119 "the record carrying its cron pairing",
8120 )
8121 .await;
8122
8123 to_server
8124 .send(Message::Text(
8125 serde_json::to_string(&ClientMessage::More {
8126 conversation: slot.clone(),
8127 before: None,
8128 turns: 5,
8129 })
8130 .expect("the ask encodes")
8131 .into(),
8132 ))
8133 .expect("the ask goes");
8134 until(
8135 || {
8136 written.lock().expect("the frame log").iter().any(|frame| {
8137 matches!(
8138 frame,
8139 ServerMessage::Page { cron_names, .. }
8140 if cron_names
8141 .get("p-cron")
8142 .and_then(|name| name.description.as_deref())
8143 == Some("Morning summary")
8144 )
8145 })
8146 },
8147 "the page carrying its cron pairing",
8148 )
8149 .await;
8150
8151 to_server
8155 .send(Message::Text(
8156 serde_json::to_string(&ClientMessage::More {
8157 conversation: slot.clone(),
8158 before: Some(below_the_second_row),
8159 turns: 5,
8160 })
8161 .expect("the ask encodes")
8162 .into(),
8163 ))
8164 .expect("the walk's ask goes");
8165 until(
8166 || {
8167 let frames = written.lock().expect("the frame log");
8168 let pages: Vec<&ServerMessage> = frames
8169 .iter()
8170 .filter(|frame| matches!(frame, ServerMessage::Page { .. }))
8171 .collect();
8172 pages.len() >= 2
8173 && matches!(
8174 pages.last(),
8175 Some(ServerMessage::Page { turns, cron_names, .. })
8176 if cron_names
8177 .get("p-cron")
8178 .and_then(|name| name.description.as_deref())
8179 == Some("Morning summary")
8180 && turns.iter().flat_map(|turn| turn.messages.iter()).any(|frame| {
8181 frame.get("uuid").and_then(serde_json::Value::as_str)
8182 == Some("p-cron")
8183 })
8184 )
8185 },
8186 "the below-floor page carrying its cron pairing, with the fired row in its span",
8187 )
8188 .await;
8189
8190 drop(to_server);
8191 let _ = driven.await;
8192 }
8193
8194 #[tokio::test]
8197 async fn subscribing_a_seat_does_not_attach_it_and_showing_it_does() {
8198 let fleet =
8199 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8200 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8201 let slot = lead();
8202 let Driven { to_server, written, driven, .. } = connection(&state);
8203
8204 subscribed(&to_server, &slot);
8205 until(
8206 || {
8207 written
8208 .lock()
8209 .expect("the frame log")
8210 .iter()
8211 .any(|frame| matches!(frame, ServerMessage::Snapshot { .. }))
8212 },
8213 "the subscribe's snapshot",
8214 )
8215 .await;
8216 assert!(
8217 !Live::lock(&state.live).is_attached(&slot),
8218 "a subscribe attached the seat, so it spends marks nobody looked at",
8219 );
8220
8221 showing(&to_server, Some(slot.clone()));
8222 until(|| Live::lock(&state.live).is_attached(&slot), "showing the seat to attach it").await;
8223
8224 showing(&to_server, None);
8225 until(|| !Live::lock(&state.live).is_attached(&slot), "showing nothing to detach it").await;
8226
8227 drop(to_server);
8228 let _ = driven.await;
8229 }
8230
8231 #[tokio::test]
8235 async fn showing_the_same_seat_twice_is_a_no_op() {
8236 let fleet =
8237 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8238 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8239 let slot = lead();
8240 let Driven { to_server, written, driven, .. } = connection(&state);
8241 subscribed_home(&to_server, &written).await;
8242 subscribed(&to_server, &slot);
8243 until(|| seat_read(&written), "the seat's snapshot").await;
8244
8245 showing(&to_server, Some(slot.clone()));
8246 until(
8247 || Live::lock(&state.live).is_attached(&slot),
8248 "the first showing to attach the seat",
8249 )
8250 .await;
8251
8252 fleet.workspace().record_death_for_test(
8260 &slot,
8261 "dead-occupant",
8262 "the claude subprocess closed its output stream",
8263 );
8264
8265 showing(&to_server, Some(slot.clone()));
8266 subscribed_home_again(&to_server, &written).await;
8270 assert_eq!(
8271 home_snapshots(&written),
8272 2,
8273 "a repeat showing spent the failure mark and re-encoded the home",
8274 );
8275 assert!(Live::lock(&state.live).is_attached(&slot), "the seat was detached");
8276
8277 drop(to_server);
8278 let _ = driven.await;
8279 }
8280
8281 #[tokio::test]
8286 async fn a_showing_before_its_subscribe_spends_when_the_subscribe_lands() {
8287 let fleet =
8288 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8289 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8290 let slot = lead();
8291 fleet.install_agent("TestOrg", "proj", "lead");
8295 let Driven { to_server, written, driven, .. } = connection(&state);
8296 subscribed_home(&to_server, &written).await;
8297 arm_diamond(&state, &slot);
8298
8299 showing(&to_server, Some(slot.clone()));
8300 let barrier = subscribed_home_again(&to_server, &written).await;
8305 assert!(
8306 !Live::lock(&state.live).is_attached(&slot),
8307 "a showing for an unwatched seat attached it",
8308 );
8309 assert!(
8310 barrier["unseen"].as_array().is_some_and(|unseen| {
8311 unseen.iter().any(|held| held["label"] == "lead" && held["project"] == "proj")
8312 }),
8313 "a showing with no read served spent the diamond: {barrier}",
8314 );
8315
8316 subscribed(&to_server, &slot);
8317 until(|| seat_read(&written), "the seat's snapshot").await;
8318 until(
8319 || Live::lock(&state.live).is_attached(&slot),
8320 "the subscribe completing the pair to attach it",
8321 )
8322 .await;
8323 until(|| home_snapshots(&written) == 3, "the home refresh a spent mark owes").await;
8324
8325 drop(to_server);
8326 let _ = driven.await;
8327 }
8328
8329 #[tokio::test]
8333 async fn a_close_detaches_only_the_seat_that_connection_showed() {
8334 let fleet =
8335 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8336 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8337 let slot = lead();
8338
8339 let Driven { to_server: first, written: first_written, driven: first_driven, .. } =
8340 connection(&state);
8341 subscribed(&first, &slot);
8344 until(
8345 || {
8346 first_written
8347 .lock()
8348 .expect("the frame log")
8349 .iter()
8350 .any(|frame| matches!(frame, ServerMessage::Snapshot { .. }))
8351 },
8352 "the first connection's snapshot",
8353 )
8354 .await;
8355 showing(&first, Some(slot.clone()));
8356 until(
8357 || Live::lock(&state.live).is_attached(&slot),
8358 "the first connection to show the seat",
8359 )
8360 .await;
8361
8362 let Driven { to_server: second, written: second_written, driven: second_driven, .. } =
8363 connection(&state);
8364 subscribed(&second, &slot);
8365 until(
8366 || {
8367 second_written
8368 .lock()
8369 .expect("the frame log")
8370 .iter()
8371 .any(|frame| matches!(frame, ServerMessage::Snapshot { .. }))
8372 },
8373 "the second connection's snapshot",
8374 )
8375 .await;
8376
8377 drop(second);
8378 let _ = second_driven.await;
8379 assert!(
8380 Live::lock(&state.live).is_attached(&slot),
8381 "a subscriber's close detached a seat another connection is showing",
8382 );
8383
8384 drop(first);
8385 let _ = first_driven.await;
8386 assert!(
8387 !Live::lock(&state.live).is_attached(&slot),
8388 "the showing connection's close did not let the seat go",
8389 );
8390 }
8391
8392 #[tokio::test]
8397 async fn a_connection_that_switched_seats_does_not_detach_the_old_one_twice() {
8398 let fleet =
8399 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8400 let first = lead();
8401 let second = SessionSlot::worker("TestOrg", "proj", "w1");
8402 fleet.install_agent("TestOrg", "proj", "lead");
8407 fleet.add_worker("TestOrg", "proj", "w1").expect("the worker row");
8408 fleet.install_agent("TestOrg", "proj", "w1");
8409 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8410
8411 let Driven { to_server, written, driven, .. } = connection(&state);
8412 let Driven { to_server: other, written: other_written, driven: other_driven, .. } =
8413 connection(&state);
8414 subscribed(&to_server, &first);
8416 subscribed(&other, &first);
8417 until(
8418 || {
8419 [&written, &other_written].iter().all(|log| {
8420 log.lock()
8421 .expect("the frame log")
8422 .iter()
8423 .any(|frame| matches!(frame, ServerMessage::Snapshot { .. }))
8424 })
8425 },
8426 "both connections' snapshots",
8427 )
8428 .await;
8429 showing(&to_server, Some(first.clone()));
8430 showing(&other, Some(first.clone()));
8431 until(
8432 || Live::lock(&state.live).is_attached(&first),
8433 "both connections to show the first seat",
8434 )
8435 .await;
8436
8437 subscribed(&to_server, &second);
8438 showing(&to_server, Some(second.clone()));
8439 until(
8440 || Live::lock(&state.live).is_attached(&second),
8441 "the switch to show the second seat",
8442 )
8443 .await;
8444 assert!(
8445 Live::lock(&state.live).is_attached(&first),
8446 "the switch detached the seat the other connection still shows",
8447 );
8448
8449 drop(to_server);
8450 let _ = driven.await;
8451 assert!(
8452 Live::lock(&state.live).is_attached(&first),
8453 "the close detached the seat the connection had already left",
8454 );
8455 assert!(
8456 !Live::lock(&state.live).is_attached(&second),
8457 "the close did not let go of the seat the connection showed",
8458 );
8459
8460 drop(other);
8461 let _ = other_driven.await;
8462 }
8463
8464 #[derive(Default)]
8467 struct StalledDetach {
8468 detaches: std::sync::atomic::AtomicUsize,
8469 attach_entered: std::sync::atomic::AtomicBool,
8472 }
8473
8474 #[async_trait::async_trait]
8475 impl forge_workspace::browser::BrowserHost for StalledDetach {
8476 async fn call(
8477 &self,
8478 _seat: &SessionSlot,
8479 _tool: &str,
8480 _args: serde_json::Value,
8481 ) -> Result<Vec<forge_primitives::browser::BrowserPart>, String> {
8482 Err("no fleet here".to_owned())
8483 }
8484
8485 async fn attach(
8486 &self,
8487 _viewer: u64,
8488 _seat: &SessionSlot,
8489 _tab: Option<u16>,
8490 ) -> Result<BrowserAttachment, String> {
8491 self.attach_entered.store(true, Ordering::Relaxed);
8492 std::future::pending::<Result<BrowserAttachment, String>>().await
8493 }
8494
8495 async fn detach(&self, _viewer: u64) {
8496 self.detaches.fetch_add(1, Ordering::Relaxed);
8497 std::future::pending::<()>().await;
8500 }
8501
8502 async fn input(
8503 &self,
8504 _profile: &str,
8505 _tab: u16,
8506 _viewer: u64,
8507 _seq: u64,
8508 _event: &forge_primitives::browser::BrowserInputEvent,
8509 ) -> Result<(), String> {
8510 Err("no fleet here".to_owned())
8511 }
8512
8513 async fn answer_files(
8514 &self,
8515 _profile: &str,
8516 _tab: u16,
8517 _id: u64,
8518 _files: Vec<forge_primitives::browser::BrowserUpload>,
8519 ) -> Result<(), String> {
8520 Err("no fleet here".to_owned())
8521 }
8522
8523 async fn take_control(
8524 &self,
8525 _profile: &str,
8526 _tab: u16,
8527 _viewer: u64,
8528 ) -> Result<(), String> {
8529 Err("no fleet here".to_owned())
8530 }
8531
8532 async fn resize(
8533 &self,
8534 _profile: &str,
8535 _tab: u16,
8536 _viewer: u64,
8537 _width: u32,
8538 _height: u32,
8539 _scale: f64,
8540 ) {
8541 }
8542
8543 async fn release_control(&self, _profile: &str, _tab: u16, _viewer: u64) {}
8544
8545 async fn nav(
8546 &self,
8547 _profile: &str,
8548 _tab: u16,
8549 _viewer: u64,
8550 _back: bool,
8551 ) -> Result<forge_primitives::browser::BrowserNavState, String> {
8552 Err("no fleet here".to_owned())
8553 }
8554
8555 async fn nav_state(
8556 &self,
8557 _profile: &str,
8558 _tab: u16,
8559 ) -> Option<forge_primitives::browser::BrowserNavState> {
8560 None
8561 }
8562
8563 async fn tabs(
8564 &self,
8565 _seat: &SessionSlot,
8566 ) -> Result<forge_primitives::browser::BrowserTabList, String> {
8567 Err("no fleet here".to_owned())
8568 }
8569
8570 async fn tab_stream(
8571 &self,
8572 _seat: &SessionSlot,
8573 ) -> Result<tokio::sync::watch::Receiver<forge_primitives::browser::BrowserTabList>, String>
8574 {
8575 Err("no fleet here".to_owned())
8576 }
8577
8578 async fn sweep_sessions(&self, _seats: &[SessionSlot], _generation: u64) {}
8579
8580 async fn replace_seat(&self, _seat: &SessionSlot) {}
8581
8582 fn session_ended(&self, _seat: &SessionSlot) {}
8583
8584 async fn hand_off_started(
8585 &self,
8586 _seat: &SessionSlot,
8587 _target: &str,
8588 ) -> Result<u16, String> {
8589 Err("no fleet here".to_owned())
8590 }
8591
8592 async fn hand_off_finished(&self, _seat: &SessionSlot, _tab: u16) {}
8593
8594 async fn warm(&self, _seat: &SessionSlot) {}
8595
8596 async fn stop(&self) -> Vec<String> {
8597 Vec::new()
8598 }
8599 }
8600
8601 #[test]
8606 fn an_ending_names_how_the_connection_went() {
8607 assert_eq!(ending(&Ok(())), "closed");
8608 assert_eq!(ending(&Err(anyhow::Error::new(PeerVanished))), "vanished");
8609 assert_eq!(
8610 ending(&Err(anyhow::Error::new(WedgeStalled {
8611 bound: WRITE_BOUND,
8612 silent: WRITE_BOUND,
8613 }))),
8614 "wedged",
8615 );
8616 assert_eq!(ending(&Err(anyhow::anyhow!("a socket error"))), "error");
8617 }
8618
8619 #[tokio::test(start_paused = true)]
8627 async fn a_wedged_connection_whose_detach_stalls_still_ends() {
8628 let fleet =
8629 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8630 let host = Arc::new(StalledDetach::default());
8631 fleet.workspace().install_browser_host(
8632 Arc::clone(&host) as Arc<dyn forge_workspace::browser::BrowserHost>
8633 );
8634 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()).with_heartbeat(
8635 crate::transport::Heartbeat {
8636 every: Duration::from_millis(20),
8637 dead: Duration::from_secs(30),
8638 wedge: Duration::from_millis(200),
8639 },
8640 ));
8641 let Driven { to_server, driven, .. } = connection(&state);
8642
8643 to_server
8645 .send(Message::Text(
8646 serde_json::to_string(&ClientMessage::BrowserView {
8647 seat: lead(),
8648 tab: None,
8649 reply_to: None,
8650 })
8651 .expect("the attach encodes")
8652 .into(),
8653 ))
8654 .expect("the message goes");
8655 let mut turns = 0;
8656 while !host.attach_entered.load(Ordering::Relaxed) {
8657 assert!(turns < 100_000, "the attach never parked");
8658 turns += 1;
8659 tokio::task::yield_now().await;
8660 }
8661 drop(to_server);
8662
8663 let ended = tokio::time::timeout(WRITE_BOUND * 2, driven)
8664 .await
8665 .expect("a stalled detach held a wedged connection open past its bound")
8666 .expect("the connection task joins");
8667 let error = ended.expect_err("a wedged connection did not end cleanly");
8668 assert!(
8669 error.downcast_ref::<WedgeStalled>().is_some(),
8670 "the connection ended for another reason: {error:?}",
8671 );
8672 assert_eq!(
8673 host.detaches.load(Ordering::Relaxed),
8674 1,
8675 "the teardown never attempted the detach",
8676 );
8677 }
8678
8679 #[tokio::test(start_paused = true)]
8684 async fn only_a_completed_flush_moves_the_watchdog_clock() {
8685 let progress = Arc::new(Progress::new());
8686 let socket =
8687 ClientSocket::new(YieldsOnce { yielded: false }, false).watched(Arc::clone(&progress));
8688 let mut socket = std::pin::pin!(socket);
8689 let waker = futures_util::task::noop_waker();
8690 let mut cx = std::task::Context::from_waker(&waker);
8691
8692 tokio::time::advance(Duration::from_secs(10)).await;
8693 let parked = progress.silent_for(tokio::time::Instant::now());
8694 assert_eq!(parked, Duration::from_secs(10), "nothing has flushed yet");
8695
8696 assert!(
8697 futures_util::Sink::poll_flush(socket.as_mut(), &mut cx).is_pending(),
8698 "the sink was expected to yield once",
8699 );
8700 assert_eq!(
8701 progress.silent_for(tokio::time::Instant::now()),
8702 parked,
8703 "a flush that has not completed is not progress",
8704 );
8705
8706 assert!(futures_util::Sink::poll_flush(socket.as_mut(), &mut cx).is_ready());
8707 assert_eq!(
8708 progress.silent_for(tokio::time::Instant::now()),
8709 Duration::ZERO,
8710 "a completed flush is progress",
8711 );
8712 }
8713
8714 #[tokio::test]
8718 async fn showing_a_seat_reaches_the_wedge_record() {
8719 let fleet =
8720 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8721 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8722 let slot = lead();
8723 let Driven { to_server, progress, .. } = connection(&state);
8724
8725 subscribed(&to_server, &slot);
8726 showing(&to_server, Some(slot.clone()));
8727 until(
8728 || progress.showing_seat().as_ref() == Some(&slot),
8729 "the showing reached the wedge record",
8730 )
8731 .await;
8732 }
8733
8734 #[tokio::test]
8737 async fn showing_a_spent_seat_refreshes_the_home_and_subscribing_does_not() {
8738 let fleet =
8739 crate::testing::Fleet::new(&[("TestOrg", &["proj"])]).expect("the fleet builds");
8740 let state = Arc::new(TransportState::for_workspace(&fleet.workspace()));
8741 let slot = lead();
8742 let Driven { to_server, written, driven, .. } = connection(&state);
8743
8744 to_server
8745 .send(Message::Text(
8746 serde_json::to_string(&ClientMessage::Subscribe {
8747 what: Subject::Home,
8748 answering: false,
8749 review: false,
8750 })
8751 .expect("the subscribe encodes")
8752 .into(),
8753 ))
8754 .expect("the message goes");
8755 until(|| home_snapshots(&written) == 1, "the home's first read").await;
8756
8757 let result = serde_json::from_value(serde_json::json!({
8759 "type": "result",
8760 "subtype": "success",
8761 "duration_ms": 1,
8762 "duration_api_ms": 1,
8763 "is_error": false,
8764 "num_turns": 1,
8765 "session_id": "s",
8766 }))
8767 .expect("a result message");
8768 Live::lock(&state.live).apply(&SessionUpdate::ChatAppended {
8769 key: slot.clone(),
8770 msg: result,
8771 source: None,
8772 });
8773
8774 subscribed(&to_server, &slot);
8775 until(
8776 || {
8777 written
8778 .lock()
8779 .expect("the frame log")
8780 .iter()
8781 .any(|frame| matches!(frame, ServerMessage::Snapshot { .. }))
8782 },
8783 "the seat's snapshot",
8784 )
8785 .await;
8786 assert_eq!(home_snapshots(&written), 1, "a subscribe refreshed the home");
8787
8788 showing(&to_server, Some(slot.clone()));
8789 until(|| home_snapshots(&written) == 2, "the home refresh a spent mark owes").await;
8790
8791 drop(to_server);
8792 let _ = driven.await;
8793 }
8794
8795 #[tokio::test]
8799 async fn a_closed_tab_names_itself_in_the_failure() {
8800 use forge_primitives::browser::{BrowserTabList, BrowserViewState};
8801
8802 let (_frames, frames_rx) = tokio::sync::watch::channel(None);
8803 let (view, view_rx) = tokio::sync::watch::channel(BrowserViewState::default());
8804 drop(view);
8805 let (_nav, nav_rx) = tokio::sync::watch::channel(None);
8806 let (_failures, failures_rx) = tokio::sync::watch::channel(None::<String>);
8807 let (_files, files_rx) = tokio::sync::watch::channel(None);
8808 let (_downloads, downloads_rx) = tokio::sync::broadcast::channel(4);
8809 let (_audio, audio_rx) = tokio::sync::broadcast::channel(4);
8810 let (_tabs, tabs_rx) = tokio::sync::watch::channel(BrowserTabList {
8811 profile: "O/default".to_owned(),
8812 id: 1,
8813 up: true,
8814 connected: true,
8815 closing_at_ms: None,
8816 rows: Vec::new(),
8817 });
8818 let attachment = forge_workspace::browser::BrowserAttachment {
8819 key: "O/default".to_owned(),
8820 id: 1,
8821 tab: 3,
8822 frames: frames_rx,
8823 view: view_rx,
8824 tabs: tabs_rx,
8825 nav: nav_rx,
8826 failures: failures_rx,
8827 files: files_rx,
8828 downloads: downloads_rx,
8829 audio: audio_rx,
8830 };
8831 let mut viewing = Some(Viewing::new(attachment));
8832
8833 match next_view_event(&mut viewing).await {
8834 ViewEvent::Failed(reason) => assert_eq!(reason, "tab 3 was closed"),
8835 _ => panic!("the tab's own end is the surface's refusal"),
8836 }
8837 }
8838
8839 #[tokio::test]
8848 async fn a_viewer_that_fell_behind_the_downloads_is_told() {
8849 use forge_primitives::browser::{BrowserDownload, BrowserTabList, BrowserViewState};
8850
8851 let (downloads, downloads_rx) = tokio::sync::broadcast::channel(1);
8852 let (_frames, frames_rx) = tokio::sync::watch::channel(None);
8853 let (_view, view_rx) = tokio::sync::watch::channel(BrowserViewState::default());
8854 let (_nav, nav_rx) = tokio::sync::watch::channel(None);
8855 let (_failures, failures_rx) = tokio::sync::watch::channel(None::<String>);
8856 let (_files, files_rx) = tokio::sync::watch::channel(None);
8857 let (_audio, audio_rx) = tokio::sync::broadcast::channel(4);
8858 let (_tabs, tabs_rx) = tokio::sync::watch::channel(BrowserTabList {
8859 profile: "O/default".to_owned(),
8860 id: 1,
8861 up: true,
8862 connected: true,
8863 closing_at_ms: None,
8864 rows: Vec::new(),
8865 });
8866 let attachment = forge_workspace::browser::BrowserAttachment {
8867 key: "O/default".to_owned(),
8868 id: 1,
8869 tab: 3,
8870 frames: frames_rx,
8871 view: view_rx,
8872 tabs: tabs_rx,
8873 nav: nav_rx,
8874 failures: failures_rx,
8875 files: files_rx,
8876 downloads: downloads_rx,
8877 audio: audio_rx,
8878 };
8879 let mut viewing = Some(Viewing::new(attachment));
8880 for n in 0..2 {
8882 let offer = BrowserDownload {
8883 tab: Some(3),
8884 name: format!("file-{n}"),
8885 bytes: Vec::new(),
8886 error: None,
8887 };
8888 downloads.send(offer).expect("the viewer is attached");
8889 }
8890
8891 match next_view_event(&mut viewing).await {
8892 ViewEvent::Download { download, dropped, lost } => {
8893 assert!(
8894 dropped,
8895 "the shell is told this is a dropped offer, so it can title it as a loss",
8896 );
8897 assert_eq!(
8898 download.tab, None,
8899 "the loss is the profile's, so the notice names no tab",
8900 );
8901 assert_eq!(lost, Some(1), "and the count the title is read from");
8902 let why = download.error.expect("a lag is answered with its own reason");
8903 assert_eq!(
8904 why,
8905 "a finished download was dropped before this viewer read it: the profile's \
8906 backlog holds at most four finished downloads at a time",
8907 "the words name one profile-level loss in plain words",
8908 );
8909 assert!(download.bytes.is_empty(), "there are no bytes to save");
8910 }
8911 _ => panic!("a lagged download is answered as a download with a reason"),
8912 }
8913 }
8914
8915 #[tokio::test]
8919 async fn a_viewer_two_offers_behind_hears_the_count() {
8920 use forge_primitives::browser::{BrowserDownload, BrowserTabList, BrowserViewState};
8921
8922 let (downloads, downloads_rx) = tokio::sync::broadcast::channel(1);
8923 let (_frames, frames_rx) = tokio::sync::watch::channel(None);
8924 let (_view, view_rx) = tokio::sync::watch::channel(BrowserViewState::default());
8925 let (_nav, nav_rx) = tokio::sync::watch::channel(None);
8926 let (_failures, failures_rx) = tokio::sync::watch::channel(None::<String>);
8927 let (_files, files_rx) = tokio::sync::watch::channel(None);
8928 let (_audio, audio_rx) = tokio::sync::broadcast::channel(4);
8929 let (_tabs, tabs_rx) = tokio::sync::watch::channel(BrowserTabList {
8930 profile: "O/default".to_owned(),
8931 id: 1,
8932 up: true,
8933 connected: true,
8934 closing_at_ms: None,
8935 rows: Vec::new(),
8936 });
8937 let attachment = forge_workspace::browser::BrowserAttachment {
8938 key: "O/default".to_owned(),
8939 id: 1,
8940 tab: 3,
8941 frames: frames_rx,
8942 view: view_rx,
8943 tabs: tabs_rx,
8944 nav: nav_rx,
8945 failures: failures_rx,
8946 files: files_rx,
8947 downloads: downloads_rx,
8948 audio: audio_rx,
8949 };
8950 let mut viewing = Some(Viewing::new(attachment));
8951 for n in 0..3 {
8952 let offer = BrowserDownload {
8953 tab: Some(3),
8954 name: format!("file-{n}"),
8955 bytes: Vec::new(),
8956 error: None,
8957 };
8958 downloads.send(offer).expect("the viewer is attached");
8959 }
8960
8961 match next_view_event(&mut viewing).await {
8962 ViewEvent::Download { download, lost, .. } => {
8963 assert_eq!(lost, Some(2), "two offers went before the viewer read one");
8964 assert_eq!(
8965 download.error.as_deref(),
8966 Some(
8967 "2 finished downloads were dropped before this viewer read them: the \
8968 profile's backlog holds at most four finished downloads at a time"
8969 ),
8970 "the counted body says the same number and the same ceiling",
8971 );
8972 }
8973 _ => panic!("a lagged download is answered as a download with a reason"),
8974 }
8975 }
8976}