merge(domain,infrastructure,web-server,app-tauri,frontend): fix #360 — TUI native noire (QA verte)

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<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);

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<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.

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<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).**

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<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();

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({

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<Agent[]> {
     return invoke<Agent[]>("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).

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,
     });

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,
       },