merge(domain,infrastructure,web-server,app-tauri,frontend): fix #360 — TUI native noire (QA verte)
This commit is contained in:
@ -1318,8 +1318,14 @@ pub async fn open_terminal(
|
|||||||
// stream is a blocking iterator, so it runs on a dedicated OS thread; it
|
// 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.
|
// ends when the PTY hits EOF (process exit) or this attach is superseded.
|
||||||
let handle = PtyHandle { session_id };
|
let handle = PtyHandle { session_id };
|
||||||
match state.pty_port.subscribe_output(&handle) {
|
match state.pty_port.subscribe_output_with_scrollback(&handle) {
|
||||||
Ok(stream) => {
|
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);
|
let bridge: std::sync::Arc<PtyBridge> = std::sync::Arc::clone(&state.pty_bridge);
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
for chunk in stream {
|
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
|
// 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.
|
// ends when the PTY hits EOF or this attach is superseded.
|
||||||
let handle = PtyHandle { session_id };
|
let handle = PtyHandle { session_id };
|
||||||
match state.pty_port.subscribe_output(&handle) {
|
match state.pty_port.subscribe_output_with_scrollback(&handle) {
|
||||||
Ok(stream) => {
|
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);
|
let bridge: std::sync::Arc<PtyBridge> = std::sync::Arc::clone(&state.pty_bridge);
|
||||||
// Captured for the level-2 tap (moved into the pump thread).
|
// Captured for the level-2 tap (moved into the pump thread).
|
||||||
let service = std::sync::Arc::clone(&state.session_limit_service);
|
let service = std::sync::Arc::clone(&state.session_limit_service);
|
||||||
|
|||||||
@ -1749,6 +1749,26 @@ pub trait PtyPort: Send + Sync {
|
|||||||
/// [`PtyError`] if the handle is unknown.
|
/// [`PtyError`] if the handle is unknown.
|
||||||
fn subscribe_output(&self, handle: &PtyHandle) -> Result<OutputStream, PtyError>;
|
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
|
/// Returns the recent output retained for the session (a bounded scrollback
|
||||||
/// ring buffer, the most recent bytes). Used to repaint a view that
|
/// ring buffer, the most recent bytes). Used to repaint a view that
|
||||||
/// re-attaches to a live PTY so the terminal isn't blank.
|
/// re-attaches to a live PTY so the terminal isn't blank.
|
||||||
|
|||||||
@ -113,6 +113,15 @@ impl Broadcast {
|
|||||||
rx
|
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.
|
/// Returns the retained scrollback as a contiguous byte vector.
|
||||||
fn snapshot(&self) -> Vec<u8> {
|
fn snapshot(&self) -> Vec<u8> {
|
||||||
self.scrollback.iter().copied().collect()
|
self.scrollback.iter().copied().collect()
|
||||||
@ -450,6 +459,23 @@ impl PtyPort for PortablePtyAdapter {
|
|||||||
Ok(Box::new(rx.into_iter()))
|
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> {
|
fn scrollback(&self, handle: &PtyHandle) -> Result<Vec<u8>, PtyError> {
|
||||||
let map = self
|
let map = self
|
||||||
.sessions
|
.sessions
|
||||||
@ -555,6 +581,18 @@ mod tests {
|
|||||||
let _second = hub.subscribe();
|
let _second = hub.subscribe();
|
||||||
assert_eq!(hub.snapshot(), b"abc".to_vec());
|
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).**
|
/// **End-to-end PTY × Landlock enforcement (lot LP4-3).**
|
||||||
|
|||||||
@ -1565,7 +1565,7 @@ async fn ws_open_terminal(
|
|||||||
code: "WS_CLOSED".to_owned(),
|
code: "WS_CLOSED".to_owned(),
|
||||||
message: "websocket output 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(
|
async fn ws_launch_agent(
|
||||||
@ -1626,7 +1626,7 @@ async fn send_launch_agent_attached(
|
|||||||
code: "WS_CLOSED".to_owned(),
|
code: "WS_CLOSED".to_owned(),
|
||||||
message: "websocket output 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(
|
async fn execute_launch_agent_for_ws(
|
||||||
@ -1764,7 +1764,7 @@ async fn ws_attach_terminal(
|
|||||||
message: "websocket output closed".to_owned(),
|
message: "websocket output closed".to_owned(),
|
||||||
})?;
|
})?;
|
||||||
let sink = WsPtySink::new(sid, tx.clone(), next_seq);
|
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(
|
fn attach_sink_and_pump(
|
||||||
@ -1772,17 +1772,27 @@ fn attach_sink_and_pump(
|
|||||||
sid: SessionId,
|
sid: SessionId,
|
||||||
sink: WsPtySink,
|
sink: WsPtySink,
|
||||||
owned: &Arc<Mutex<Vec<WsAttachment>>>,
|
owned: &Arc<Mutex<Vec<WsAttachment>>>,
|
||||||
|
replay_initial_scrollback: bool,
|
||||||
) -> Result<(), ErrorDto> {
|
) -> 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));
|
let gen = state.ws_pty_bridge.register(sid, Arc::new(sink));
|
||||||
if let Ok(mut owned) = owned.lock() {
|
if let Ok(mut owned) = owned.lock() {
|
||||||
owned.push(WsAttachment::Pty { session: sid, gen });
|
owned.push(WsAttachment::Pty { session: sid, gen });
|
||||||
}
|
}
|
||||||
let handle = PtyHandle { session_id: sid };
|
if !scrollback.is_empty() {
|
||||||
let stream = state
|
let _ = state.ws_pty_bridge.send_output(&sid, scrollback);
|
||||||
.app
|
}
|
||||||
.pty_port
|
|
||||||
.subscribe_output(&handle)
|
|
||||||
.map_err(|err| ErrorDto::from(application::AppError::from(err)))?;
|
|
||||||
let bridge = Arc::clone(&state.ws_pty_bridge);
|
let bridge = Arc::clone(&state.ws_pty_bridge);
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
for chunk in stream {
|
for chunk in stream {
|
||||||
@ -8112,6 +8122,40 @@ mod tests {
|
|||||||
close_ws_session(&state, &session_id).await;
|
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]
|
#[tokio::test]
|
||||||
async fn websocket_attach_terminal_replays_scrollback_in_attached_ack() {
|
async fn websocket_attach_terminal_replays_scrollback_in_attached_ack() {
|
||||||
let state = state();
|
let state = state();
|
||||||
|
|||||||
@ -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 () => {
|
it("launchAgentChat rejects a PTY-routed launch before returning a fake structured session", async () => {
|
||||||
const log = vi.spyOn(console, "error").mockImplementation(() => {});
|
const log = vi.spyOn(console, "error").mockImplementation(() => {});
|
||||||
invoke.mockResolvedValueOnce({
|
invoke.mockResolvedValueOnce({
|
||||||
|
|||||||
@ -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 {
|
export class TauriAgentGateway implements AgentGateway {
|
||||||
listAgents(projectId: string): Promise<Agent[]> {
|
listAgents(projectId: string): Promise<Agent[]> {
|
||||||
return invoke<Agent[]>("list_agents", { projectId });
|
return invoke<Agent[]>("list_agents", { projectId });
|
||||||
@ -214,6 +230,11 @@ export class TauriAgentGateway implements AgentGateway {
|
|||||||
agentId,
|
agentId,
|
||||||
rows: options.rows,
|
rows: options.rows,
|
||||||
cols: options.cols,
|
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
|
// 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.
|
// backend resumes it; absent ⇒ a fresh cell that may get a new id.
|
||||||
conversationId: options.conversationId ?? null,
|
conversationId: options.conversationId ?? null,
|
||||||
@ -224,6 +245,10 @@ export class TauriAgentGateway implements AgentGateway {
|
|||||||
onOutput: channel,
|
onOutput: channel,
|
||||||
});
|
});
|
||||||
|
|
||||||
|
if (res.cellKind !== "pty") {
|
||||||
|
throw nativeLaunchRoutedToNonPtyError(projectId, agentId, options, res);
|
||||||
|
}
|
||||||
|
|
||||||
const base = makeTerminalHandle(res.sessionId, channel);
|
const base = makeTerminalHandle(res.sessionId, channel);
|
||||||
// Surface the id assigned by this launch (so the caller persists it on the
|
// Surface the id assigned by this launch (so the caller persists it on the
|
||||||
// leaf and resumes next time).
|
// leaf and resumes next time).
|
||||||
|
|||||||
@ -220,6 +220,7 @@ describe("HttpAgentGateway WS round-trip (B6 frames)", () => {
|
|||||||
nodeId: "node-1",
|
nodeId: "node-1",
|
||||||
rows: 24,
|
rows: 24,
|
||||||
cols: 80,
|
cols: 80,
|
||||||
|
cellKind: "pty",
|
||||||
conversationId: "resume-9",
|
conversationId: "resume-9",
|
||||||
freshStart: false,
|
freshStart: false,
|
||||||
});
|
});
|
||||||
|
|||||||
@ -416,6 +416,10 @@ export class WsLiveClient {
|
|||||||
nodeId: params.nodeId ?? null,
|
nodeId: params.nodeId ?? null,
|
||||||
rows: params.rows,
|
rows: params.rows,
|
||||||
cols: params.cols,
|
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,
|
conversationId: params.conversationId ?? null,
|
||||||
freshStart: params.freshStart ?? false,
|
freshStart: params.freshStart ?? false,
|
||||||
},
|
},
|
||||||
|
|||||||
Reference in New Issue
Block a user