fix(domain,infrastructure,web-server,app-tauri,frontend): TUI native noire — route PTY explicite et replay du scrollback pré-attach sans perte — #360 (QA verte)

This commit is contained in:
2026-09-16 19:08:01 +02:00
parent a833af0fb5
commit 355a482fee
8 changed files with 220 additions and 13 deletions

View File

@ -1318,8 +1318,14 @@ pub async fn open_terminal(
// stream is a blocking iterator, so it runs on a dedicated OS thread; it
// ends when the PTY hits EOF (process exit) or this attach is superseded.
let handle = PtyHandle { session_id };
match state.pty_port.subscribe_output(&handle) {
Ok(stream) => {
match state.pty_port.subscribe_output_with_scrollback(&handle) {
Ok((scrollback, stream)) => {
// The child can paint its initial screen before this command has
// finished wiring the Tauri channel. Replay that retained prefix
// before the live pump so a one-shot TUI draw is never lost.
if !scrollback.is_empty() {
let _ = state.pty_bridge.send_output(&session_id, scrollback);
}
let bridge: std::sync::Arc<PtyBridge> = std::sync::Arc::clone(&state.pty_bridge);
std::thread::spawn(move || {
for chunk in stream {
@ -2654,8 +2660,14 @@ pub async fn launch_agent(
// The stream is a blocking iterator; it runs on a dedicated OS thread and
// ends when the PTY hits EOF or this attach is superseded.
let handle = PtyHandle { session_id };
match state.pty_port.subscribe_output(&handle) {
Ok(stream) => {
match state.pty_port.subscribe_output_with_scrollback(&handle) {
Ok((scrollback, stream)) => {
// Launching a CLI performs substantial setup after the process
// starts. Its first full-screen draw may therefore predate the
// frontend channel; replay the atomic pre-attach snapshot first.
if !scrollback.is_empty() {
let _ = state.pty_bridge.send_output(&session_id, scrollback);
}
let bridge: std::sync::Arc<PtyBridge> = std::sync::Arc::clone(&state.pty_bridge);
// Captured for the level-2 tap (moved into the pump thread).
let service = std::sync::Arc::clone(&state.session_limit_service);

View File

@ -1749,6 +1749,26 @@ pub trait PtyPort: Send + Sync {
/// [`PtyError`] if the handle is unknown.
fn subscribe_output(&self, handle: &PtyHandle) -> Result<OutputStream, PtyError>;
/// Atomically snapshots the retained output and subscribes to future chunks.
///
/// Presentation adapters use this when they first attach a view after spawn:
/// bytes produced before the channel was ready are returned in the snapshot,
/// while bytes produced afterwards enter the stream exactly once. Concrete
/// adapters should override this operation when snapshot/subscription share
/// a synchronization boundary. The default keeps existing port adapters
/// source-compatible, but cannot close a race between the two operations.
///
/// # Errors
/// [`PtyError`] if the handle is unknown.
fn subscribe_output_with_scrollback(
&self,
handle: &PtyHandle,
) -> Result<(Vec<u8>, OutputStream), PtyError> {
let scrollback = self.scrollback(handle)?;
let stream = self.subscribe_output(handle)?;
Ok((scrollback, stream))
}
/// Returns the recent output retained for the session (a bounded scrollback
/// ring buffer, the most recent bytes). Used to repaint a view that
/// re-attaches to a live PTY so the terminal isn't blank.

View File

@ -113,6 +113,15 @@ impl Broadcast {
rx
}
/// Captures all output produced before attachment and installs the live
/// subscriber under the same lock. This is the no-gap handoff between the
/// retained history and future chunks.
fn snapshot_and_subscribe(&mut self) -> (Vec<u8>, Receiver<Vec<u8>>) {
let snapshot = self.snapshot();
let receiver = self.subscribe();
(snapshot, receiver)
}
/// Returns the retained scrollback as a contiguous byte vector.
fn snapshot(&self) -> Vec<u8> {
self.scrollback.iter().copied().collect()
@ -450,6 +459,23 @@ impl PtyPort for PortablePtyAdapter {
Ok(Box::new(rx.into_iter()))
}
fn subscribe_output_with_scrollback(
&self,
handle: &PtyHandle,
) -> Result<(Vec<u8>, OutputStream), PtyError> {
let map = self
.sessions
.lock()
.map_err(|_| PtyError::Io("pty registry poisoned".to_owned()))?;
let live = map.get(&handle.session_id).ok_or(PtyError::NotFound)?;
let (scrollback, rx) = live
.output
.lock()
.map_err(|_| PtyError::Io("pty output hub poisoned".to_owned()))?
.snapshot_and_subscribe();
Ok((scrollback, Box::new(rx.into_iter())))
}
fn scrollback(&self, handle: &PtyHandle) -> Result<Vec<u8>, PtyError> {
let map = self
.sessions
@ -555,6 +581,18 @@ mod tests {
let _second = hub.subscribe();
assert_eq!(hub.snapshot(), b"abc".to_vec());
}
#[test]
fn snapshot_and_subscribe_hands_off_scrollback_to_live_without_duplication() {
let mut hub = Broadcast::default();
hub.push(b"before");
let (snapshot, live) = hub.snapshot_and_subscribe();
hub.push(b"after");
assert_eq!(snapshot, b"before".to_vec());
assert_eq!(live.recv().unwrap(), b"after".to_vec());
}
}
/// **End-to-end PTY × Landlock enforcement (lot LP4-3).**

View File

@ -1565,7 +1565,7 @@ async fn ws_open_terminal(
code: "WS_CLOSED".to_owned(),
message: "websocket output closed".to_owned(),
})?;
attach_sink_and_pump(state, sid, sink, owned)
attach_sink_and_pump(state, sid, sink, owned, true)
}
async fn ws_launch_agent(
@ -1626,7 +1626,7 @@ async fn send_launch_agent_attached(
code: "WS_CLOSED".to_owned(),
message: "websocket output closed".to_owned(),
})?;
attach_sink_and_pump(state, session_id, sink, owned)
attach_sink_and_pump(state, session_id, sink, owned, true)
}
async fn execute_launch_agent_for_ws(
@ -1764,7 +1764,7 @@ async fn ws_attach_terminal(
message: "websocket output closed".to_owned(),
})?;
let sink = WsPtySink::new(sid, tx.clone(), next_seq);
attach_sink_and_pump(state, sid, sink, owned)
attach_sink_and_pump(state, sid, sink, owned, false)
}
fn attach_sink_and_pump(
@ -1772,17 +1772,27 @@ fn attach_sink_and_pump(
sid: SessionId,
sink: WsPtySink,
owned: &Arc<Mutex<Vec<WsAttachment>>>,
replay_initial_scrollback: bool,
) -> Result<(), ErrorDto> {
let handle = PtyHandle { session_id: sid };
let (scrollback, stream) = if replay_initial_scrollback {
state.app.pty_port.subscribe_output_with_scrollback(&handle)
} else {
state
.app
.pty_port
.subscribe_output(&handle)
.map(|stream| (Vec::new(), stream))
}
.map_err(|err| ErrorDto::from(application::AppError::from(err)))?;
let gen = state.ws_pty_bridge.register(sid, Arc::new(sink));
if let Ok(mut owned) = owned.lock() {
owned.push(WsAttachment::Pty { session: sid, gen });
}
let handle = PtyHandle { session_id: sid };
let stream = state
.app
.pty_port
.subscribe_output(&handle)
.map_err(|err| ErrorDto::from(application::AppError::from(err)))?;
if !scrollback.is_empty() {
let _ = state.ws_pty_bridge.send_output(&sid, scrollback);
}
let bridge = Arc::clone(&state.ws_pty_bridge);
std::thread::spawn(move || {
for chunk in stream {
@ -8112,6 +8122,40 @@ mod tests {
close_ws_session(&state, &session_id).await;
}
#[tokio::test]
async fn websocket_open_terminal_delivers_the_initial_process_output() {
let state = state();
let (tx, mut rx) = mpsc::channel(16);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(
ws_frame(
"open-initial-output",
"terminal.open",
open_terminal_payload("/bin/sh", vec!["-c", "printf initial-screen; sleep 30"]),
),
&state,
&tx,
&owned,
)
.await;
let attached = recv_server_frame(&mut rx).await;
let session_id = attached_session_id(&attached);
let output = recv_server_frame(&mut rx).await;
assert_eq!(attached.kind, "terminal.attached");
assert_eq!(output.kind, "terminal.output");
assert_eq!(output.payload["sessionId"], session_id);
assert_eq!(output.payload["seq"], 0);
assert_eq!(
base64_decode(output.payload["bytesBase64"].as_str().unwrap()).unwrap(),
b"initial-screen"
);
close_ws_session(&state, &session_id).await;
}
#[tokio::test]
async fn websocket_attach_terminal_replays_scrollback_in_attached_ack() {
let state = state();