From 355a482feedff72f3f28e44585095342384c43ce Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 16 Sep 2026 19:08:01 +0200 Subject: [PATCH] =?UTF-8?q?fix(domain,infrastructure,web-server,app-tauri,?= =?UTF-8?q?frontend):=20TUI=20native=20noire=20=E2=80=94=20route=20PTY=20e?= =?UTF-8?q?xplicite=20et=20replay=20du=20scrollback=20pr=C3=A9-attach=20sa?= =?UTF-8?q?ns=20perte=20=E2=80=94=20#360=20(QA=20verte)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/app-tauri/src/commands.rs | 20 ++++-- crates/domain/src/ports.rs | 20 ++++++ crates/infrastructure/src/pty/mod.rs | 38 +++++++++++ crates/web-server/src/lib.rs | 62 +++++++++++++++--- frontend/src/adapters/agent.test.ts | 63 +++++++++++++++++++ frontend/src/adapters/agent.ts | 25 ++++++++ .../src/adapters/http/agentGateway.test.ts | 1 + frontend/src/adapters/http/wsLiveClient.ts | 4 ++ 8 files changed, 220 insertions(+), 13 deletions(-) diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 01b7f1f3..9f32604a 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -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 = 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 = 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); diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index 759e4aa9..1501c541 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -1749,6 +1749,26 @@ pub trait PtyPort: Send + Sync { /// [`PtyError`] if the handle is unknown. fn subscribe_output(&self, handle: &PtyHandle) -> Result; + /// 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, 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. diff --git a/crates/infrastructure/src/pty/mod.rs b/crates/infrastructure/src/pty/mod.rs index 923693f9..5d3703e1 100644 --- a/crates/infrastructure/src/pty/mod.rs +++ b/crates/infrastructure/src/pty/mod.rs @@ -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, Receiver>) { + let snapshot = self.snapshot(); + let receiver = self.subscribe(); + (snapshot, receiver) + } + /// Returns the retained scrollback as a contiguous byte vector. fn snapshot(&self) -> Vec { 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, 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, 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).** diff --git a/crates/web-server/src/lib.rs b/crates/web-server/src/lib.rs index b4403aff..fc297962 100644 --- a/crates/web-server/src/lib.rs +++ b/crates/web-server/src/lib.rs @@ -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>>, + 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(); diff --git a/frontend/src/adapters/agent.test.ts b/frontend/src/adapters/agent.test.ts index 004927b3..cbe9dfbd 100644 --- a/frontend/src/adapters/agent.test.ts +++ b/frontend/src/adapters/agent.test.ts @@ -400,6 +400,69 @@ describe("TauriAgentGateway invoke payloads", () => { }); }); + it("launchAgent requests and accepts only the PTY route for a native TUI", async () => { + invoke.mockResolvedValueOnce({ + sessionId: "pty-session-1", + cwd: "/repo", + rows: 24, + cols: 80, + cellKind: "pty", + assignedConversationId: "conversation-1", + }); + + const out = await new TauriAgentGateway().launchAgent( + "proj-1", + "agent-2", + { + cwd: "/repo", + rows: 24, + cols: 80, + conversationId: "conversation-0", + nodeId: "node-3", + }, + vi.fn(), + ); + + expect(invoke).toHaveBeenCalledWith("launch_agent", { + request: { + projectId: "proj-1", + agentId: "agent-2", + rows: 24, + cols: 80, + cellKind: "pty", + conversationId: "conversation-0", + freshStart: false, + nodeId: "node-3", + }, + onOutput: expect.anything(), + }); + expect(out.sessionId).toBe("pty-session-1"); + expect(out.assignedConversationId).toBe("conversation-1"); + }); + + it("launchAgent rejects a chat-routed launch instead of returning a silent PTY handle", async () => { + invoke.mockResolvedValueOnce({ + sessionId: "chat-session-1", + cwd: "/repo", + rows: 24, + cols: 80, + cellKind: "chat", + }); + + await expect( + new TauriAgentGateway().launchAgent( + "proj-1", + "agent-2", + { cwd: "/repo", rows: 24, cols: 80, nodeId: "node-3" }, + vi.fn(), + ), + ).rejects.toEqual({ + code: "NATIVE_TUI_ROUTED_TO_NON_PTY", + message: + "native TUI launch for agent agent-2 in project proj-1 returned cellKind=chat; expected pty. sessionId=chat-session-1; nodeId=node-3", + }); + }); + it("launchAgentChat rejects a PTY-routed launch before returning a fake structured session", async () => { const log = vi.spyOn(console, "error").mockImplementation(() => {}); invoke.mockResolvedValueOnce({ diff --git a/frontend/src/adapters/agent.ts b/frontend/src/adapters/agent.ts index b8525b14..86a34e8f 100644 --- a/frontend/src/adapters/agent.ts +++ b/frontend/src/adapters/agent.ts @@ -74,6 +74,22 @@ function structuredLaunchRoutedToPtyError( }; } +function nativeLaunchRoutedToNonPtyError( + projectId: string, + agentId: string, + options: OpenTerminalOptions, + response: LaunchAgentResponse, +): GatewayError { + const actual = response.cellKind ?? "missing"; + return { + code: "NATIVE_TUI_ROUTED_TO_NON_PTY", + message: + `native TUI launch for agent ${agentId} in project ${projectId} ` + + `returned cellKind=${actual}; expected pty. ` + + `sessionId=${response.sessionId}; nodeId=${options.nodeId ?? "none"}`, + }; +} + export class TauriAgentGateway implements AgentGateway { listAgents(projectId: string): Promise { return invoke("list_agents", { projectId }); @@ -214,6 +230,11 @@ export class TauriAgentGateway implements AgentGateway { agentId, rows: options.rows, cols: options.cols, + // Native TUI is always the raw PTY route, including for profiles that + // also support the structured/custom CLI. Omitting this hint lets the + // backend select a chat session, whose reply stream can never feed the + // xterm output channel and leaves the cell looking blank. + cellKind: "pty", // Resume id: the leaf's persisted conversation id, when any (T4b). The // backend resumes it; absent ⇒ a fresh cell that may get a new id. conversationId: options.conversationId ?? null, @@ -224,6 +245,10 @@ export class TauriAgentGateway implements AgentGateway { onOutput: channel, }); + if (res.cellKind !== "pty") { + throw nativeLaunchRoutedToNonPtyError(projectId, agentId, options, res); + } + const base = makeTerminalHandle(res.sessionId, channel); // Surface the id assigned by this launch (so the caller persists it on the // leaf and resumes next time). diff --git a/frontend/src/adapters/http/agentGateway.test.ts b/frontend/src/adapters/http/agentGateway.test.ts index 218f0651..30c305e0 100644 --- a/frontend/src/adapters/http/agentGateway.test.ts +++ b/frontend/src/adapters/http/agentGateway.test.ts @@ -220,6 +220,7 @@ describe("HttpAgentGateway WS round-trip (B6 frames)", () => { nodeId: "node-1", rows: 24, cols: 80, + cellKind: "pty", conversationId: "resume-9", freshStart: false, }); diff --git a/frontend/src/adapters/http/wsLiveClient.ts b/frontend/src/adapters/http/wsLiveClient.ts index 961f26b2..0f522a97 100644 --- a/frontend/src/adapters/http/wsLiveClient.ts +++ b/frontend/src/adapters/http/wsLiveClient.ts @@ -416,6 +416,10 @@ export class WsLiveClient { nodeId: params.nodeId ?? null, rows: params.rows, cols: params.cols, + // The native cell must select the raw PTY route explicitly. Profiles + // may also expose a structured chat route; relying on their default can + // produce a chat session behind an xterm subscription with no output. + cellKind: "pty", conversationId: params.conversationId ?? null, freshStart: params.freshStart ?? false, },