Compare commits

..
Author SHA1 Message Date
Miguel Palhas b680266990 docs(maestro): fix Plan row rendering claim in STRUCTURED-VIEW
Plan rows are omitted from the transcript in both clients and rendered
in a separate plan rail/status line, not as a transcript checklist.
2026-09-05 10:48:15 +01:00
Miguel Palhas ded1f050b2 fix(web): omit plan transcript rows 2026-09-05 10:41:00 +01:00
Miguel Palhas a58ebfed93 Merge issue #193: preserve expanded cards on reconnect 2026-09-05 10:32:48 +01:00
Miguel Palhas 31b429a840 fix(web): preserve expanded structured cards 2026-09-05 10:29:40 +01:00
Miguel Palhas 12bd68a2af Merge issue #192: reject structured PTY operations 2026-09-05 10:24:34 +01:00
Miguel Palhas f765d415d6 fix(maestro): reject structured PTY access 2026-09-05 10:20:41 +01:00
Miguel Palhas d3e3010245 Merge issue #143: offer structured view in spawn forms 2026-09-05 09:53:21 +01:00
10 changed files with 549 additions and 56 deletions
+45 -1
View File
@@ -19,6 +19,12 @@ use tokio_tungstenite::tungstenite::Message;
pub enum AttachError {
#[error("ws: {0}")]
WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
#[error("daemon returned {status} ({code}): {message}")]
Rejected {
status: String,
code: String,
message: String,
},
#[error("io: {0}")]
Io(#[from] io::Error),
#[error("termios: {0}")]
@@ -32,6 +38,21 @@ pub enum AttachError {
}
impl AttachError {
fn from_websocket(error: tokio_tungstenite::tungstenite::Error) -> Self {
if let tokio_tungstenite::tungstenite::Error::Http(response) = &error {
if let Some(body) = response.body().as_deref() {
if let Ok(error_body) = serde_json::from_slice::<maestro_protocol::ApiError>(body) {
return Self::Rejected {
status: response.status().to_string(),
code: error_body.error.code,
message: error_body.error.message,
};
}
}
}
Self::WebSocket(error)
}
/// True when the failure was an inability to reach the WS endpoint at the
/// transport level (connection refused), as opposed to an HTTP-level
/// rejection from a daemon that did answer.
@@ -111,7 +132,9 @@ fn current_winsize() -> Option<(u16, u16)> {
/// byte that exits cleanly without killing the session.
#[allow(clippy::result_large_err)]
pub async fn run_attach_url(url: &str, detach_byte: u8) -> Result<(), AttachError> {
let (ws, _resp) = tokio_tungstenite::connect_async(url).await?;
let (ws, _resp) = tokio_tungstenite::connect_async(url)
.await
.map_err(AttachError::from_websocket)?;
let (mut sink, mut stream) = ws.split();
let raw = RawTty::enable()?;
@@ -233,4 +256,25 @@ mod tests {
Err(AttachError::InvalidDetachKey(_))
));
}
#[test]
fn attach_surfaces_structured_handshake_errors() {
let body = serde_json::to_vec(&maestro_protocol::ApiError::new(
"not_terminal",
"session is not a terminal session",
))
.unwrap();
let response = tokio_tungstenite::tungstenite::http::Response::builder()
.status(409)
.body(Some(body))
.unwrap();
let error =
AttachError::from_websocket(tokio_tungstenite::tungstenite::Error::Http(response));
assert_eq!(
error.to_string(),
"daemon returned 409 Conflict (not_terminal): session is not a terminal session"
);
}
}
+26 -6
View File
@@ -6,7 +6,7 @@ use std::io::Write;
use maestro_protocol::display::abbreviate_dir;
use maestro_protocol::{
ActivityState, ClientError, CreateSessionRequest, DaemonClient, DaemonInfo, SendInputRequest,
Session, SessionStatus, WorktreeSpec,
Session, SessionStatus, SessionView, WorktreeSpec,
};
use serde_json::Value;
@@ -179,8 +179,8 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
}
writeln!(
out,
" {:<36} {:<8} {:<9} {:<28} {:<40} {}",
"ID", "STATUS", "ACTIVITY", "DIR", "COMMAND", "BLOCKED"
" {:<36} {:<8} {:<10} {:<9} {:<28} {:<40} {}",
"ID", "STATUS", "VIEW", "ACTIVITY", "DIR", "COMMAND", "BLOCKED"
)
.ok();
for s in sorted {
@@ -190,6 +190,10 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
SessionStatus::Running => "running",
SessionStatus::Dead => "dead",
};
let view = match s.view {
SessionView::Terminal => "terminal",
SessionView::Structured => "structured",
};
let activity = match s.activity {
ActivityState::Active => "active",
ActivityState::Idle => "idle",
@@ -207,8 +211,8 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
let blocked = blocked_label(s);
writeln!(
out,
" {:<36} {:<8} {:<9} {:<28} {:<40} {blocked}",
s.id, status, activity, dir, command
" {:<36} {:<8} {:<10} {:<9} {:<28} {:<40} {blocked}",
s.id, status, view, activity, dir, command
)
.ok();
}
@@ -230,7 +234,7 @@ mod tests {
use std::path::PathBuf;
use chrono::{TimeZone, Utc};
use maestro_protocol::{ActivityState, Session, SessionStatus};
use maestro_protocol::{ActivityState, Session, SessionStatus, SessionView};
use uuid::Uuid;
use super::DaemonInfo;
@@ -286,6 +290,22 @@ mod tests {
assert!(s.contains("idle"), "{s}");
}
#[test]
fn table_exposes_terminal_and_structured_views() {
let terminal = session("/tmp", "bash", 1);
let mut structured = session("/tmp", "claude", 2);
structured.view = SessionView::Structured;
let mut buf = Vec::new();
super::print_session_table(&mut buf, &[terminal, structured], &info(""));
let out = String::from_utf8(buf).unwrap();
let lines: Vec<&str> = out.lines().collect();
let header = lines.iter().position(|line| line.contains("VIEW")).unwrap();
assert!(lines[header + 1].contains("terminal"), "{out}");
assert!(lines[header + 2].contains("structured"), "{out}");
}
#[test]
fn live_cwd_overrides_spawn_dir_in_table() {
let mut s = session("/home/naps/maestro", "bash", 1);
+19 -7
View File
@@ -26,7 +26,7 @@ use maestro_protocol::{
use serde::Deserialize;
use uuid::Uuid;
use crate::session::{QueueAcpPrompt, SessionManager};
use crate::session::{QueueAcpPrompt, SessionError, SessionManager};
use crate::sidecar::EventCursor;
#[derive(Debug, Clone)]
@@ -179,6 +179,7 @@ fn session_error_status(code: &str) -> StatusCode {
// The session outlived its pane: nothing to type into, and nothing
// the daemon can do about it.
"not_running" => StatusCode::CONFLICT,
"not_terminal" => StatusCode::CONFLICT,
"shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
_ => StatusCode::BAD_GATEWAY,
}
@@ -534,12 +535,23 @@ async fn attach_ws(
State(s): State<LocalApiState>,
ws: axum::extract::WebSocketUpgrade,
) -> Result<Response, Response> {
let Some((replay, rx)) = s.manager.subscribe_pty_with_replay(id).await else {
return Err(error_response(
StatusCode::NOT_FOUND,
"unknown_session",
"no such session on this daemon",
));
let (replay, rx) = match s.manager.subscribe_pty_with_replay(id).await {
Ok(subscription) => subscription,
Err(SessionError::NotFound(_)) => {
return Err(error_response(
StatusCode::NOT_FOUND,
"unknown_session",
"no such session on this daemon",
));
}
Err(error) => {
let info = error.to_error_info();
return Err(error_response(
session_error_status(&info.code),
&info.code,
&info.message,
));
}
};
let manager = s.manager.clone();
Ok(ws.on_upgrade(move |socket| async move {
+90 -23
View File
@@ -169,6 +169,8 @@ pub enum SessionError {
/// a stray keystroke on an attach socket, which is dropped.
#[error("session is no longer running: {0}")]
NotRunning(Uuid),
#[error("session is not a terminal session")]
NotTerminal(Uuid),
#[error("daemon is shutting down")]
ShuttingDown,
/// The kill was refused (#79): the session's worktree holds unsaved work.
@@ -205,6 +207,7 @@ impl SessionError {
SessionError::NotFound(_) => "not_found",
SessionError::InvalidEnv(_) => "invalid_env",
SessionError::NotRunning(_) => "not_running",
SessionError::NotTerminal(_) => "not_terminal",
SessionError::ShuttingDown => "shutting_down",
SessionError::KillGuarded(_) => "kill_guarded",
};
@@ -879,11 +882,17 @@ impl SessionManager {
}
}
/// Subscribe to a session's PTY broadcast. `None` if the session is
/// unknown to this daemon.
pub async fn subscribe_pty(&self, id: Uuid) -> Option<broadcast::Receiver<Vec<u8>>> {
/// Subscribe to a terminal session's PTY broadcast.
pub async fn subscribe_pty(
&self,
id: Uuid,
) -> Result<broadcast::Receiver<Vec<u8>>, SessionError> {
let sessions = self.inner.sessions.lock().await;
sessions.get(&id).map(|r| r.bridge.subscribe())
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if record.session.view != SessionView::Terminal {
return Err(SessionError::NotTerminal(id));
}
Ok(record.bridge.subscribe())
}
/// Subscribe to a session's PTY broadcast and return the replay buffer
@@ -897,11 +906,14 @@ impl SessionManager {
pub async fn subscribe_pty_with_replay(
&self,
id: Uuid,
) -> Option<(Vec<u8>, broadcast::Receiver<Vec<u8>>)> {
) -> Result<(Vec<u8>, broadcast::Receiver<Vec<u8>>), SessionError> {
let tmux = self.inner.tmux.clone();
let (result, updated) = {
let mut sessions = self.inner.sessions.lock().await;
let rec = sessions.get_mut(&id)?;
let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
if rec.session.view != SessionView::Terminal {
return Err(SessionError::NotTerminal(id));
}
rec.session.last_focused_at = Some(Utc::now());
let updated = rec.session.clone();
// For a live session, tmux's own screen beats the byte tail: the
@@ -932,7 +944,7 @@ impl SessionManager {
.inner
.events
.send(CatalogEvent::SessionUpdated { session: updated });
Some(result)
Ok(result)
}
pub async fn has(&self, id: Uuid) -> bool {
@@ -945,6 +957,15 @@ impl SessionManager {
sessions.get(&id).map(|r| r.session.status)
}
async fn terminal_status(&self, id: Uuid) -> Result<SessionStatus, SessionError> {
let sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if record.session.view != SessionView::Terminal {
return Err(SessionError::NotTerminal(id));
}
Ok(record.session.status)
}
/// Whether the natural-exit settle step (#82) has finished for this
/// session: `Some(true)` once applied or refused, `Some(false)` while
/// still pending, `None` if the session is unknown. `Dead` is observable
@@ -1626,10 +1647,9 @@ impl SessionManager {
/// its pane is gone, so this is a no-op rather than a tmux failure the
/// client would have to interpret.
pub async fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), SessionError> {
match self.status(id).await {
None => return Err(SessionError::NotFound(id)),
Some(SessionStatus::Dead) => return Ok(()),
Some(SessionStatus::Running) => {}
match self.terminal_status(id).await? {
SessionStatus::Dead => return Ok(()),
SessionStatus::Running => {}
}
let tmux = self.inner.tmux.clone();
let bytes = bytes.to_vec();
@@ -1645,10 +1665,9 @@ impl SessionManager {
/// asked for text to land somewhere, and a silent success would let a
/// script believe an agent had been told something it never heard.
pub async fn send_input(&self, id: Uuid, text: &str, submit: bool) -> Result<(), SessionError> {
match self.status(id).await {
None => return Err(SessionError::NotFound(id)),
Some(SessionStatus::Dead) => return Err(SessionError::NotRunning(id)),
Some(SessionStatus::Running) => {}
match self.terminal_status(id).await? {
SessionStatus::Dead => return Err(SessionError::NotRunning(id)),
SessionStatus::Running => {}
}
let tmux = self.inner.tmux.clone();
let bytes = text.as_bytes().to_vec();
@@ -1668,10 +1687,9 @@ impl SessionManager {
/// Resize a session's window. No-op for a dead session, as with
/// [`Self::send_keys`].
pub async fn resize(&self, id: Uuid, cols: u16, rows: u16) -> Result<(), SessionError> {
match self.status(id).await {
None => return Err(SessionError::NotFound(id)),
Some(SessionStatus::Dead) => return Ok(()),
Some(SessionStatus::Running) => {}
match self.terminal_status(id).await? {
SessionStatus::Dead => return Ok(()),
SessionStatus::Running => {}
}
let tmux = self.inner.tmux.clone();
tmux_blocking(move || tmux.resize_window(id, cols, rows))
@@ -1703,6 +1721,7 @@ impl SessionManager {
cols: u16,
rows: u16,
) -> Result<(), SessionError> {
self.terminal_status(id).await?;
let target = self
.inner
.viewports
@@ -2807,13 +2826,15 @@ mod tests {
async fn manager_with_structured_stream() -> (
tempfile::TempDir,
SessionManager,
Arc<FakePanes>,
uuid::Uuid,
Arc<tokio::sync::Mutex<AcpEventStream>>,
uuid::Uuid,
) {
let state = tempfile::tempdir().unwrap();
let panes = Arc::new(FakePanes::new());
let manager = SessionManager::new(
Arc::new(FakePanes::new()),
panes.clone(),
SidecarStore::new(state.path().join("sessions")),
WorktreeManager::new(None),
state.path().join("fifos"),
@@ -2848,12 +2869,12 @@ mod tests {
record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD);
record.acp_events = Some(events.clone());
}
(state, manager, structured.id, events, terminal.id)
(state, manager, panes, structured.id, events, terminal.id)
}
#[tokio::test]
async fn blocked_structured_activity_read_does_not_block_the_catalog() {
let (state, manager, structured_id, events, terminal_id) =
let (state, manager, _panes, structured_id, events, terminal_id) =
manager_with_structured_stream().await;
let held_stream = events.lock().await;
let activity_pass = run_structured_activity_pass(&manager.inner);
@@ -2898,9 +2919,55 @@ mod tests {
assert_eq!(structured.activity, ActivityState::Idle);
}
#[tokio::test]
async fn structured_sessions_reject_every_pty_operation_before_panes() {
let (_state, manager, panes, structured_id, _events, terminal_id) =
manager_with_structured_stream().await;
let before = panes.calls();
assert!(matches!(
manager.subscribe_pty(structured_id).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.subscribe_pty_with_replay(structured_id).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.send_keys(structured_id, b"ignored").await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.send_input(structured_id, "ignored", false).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.resize(structured_id, 80, 24).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager
.set_viewport(structured_id, manager.attach_viewport(), 80, 24)
.await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
let after = panes.calls();
assert_eq!(after, before, "structured PTY operations touched panes");
manager
.send_input(terminal_id, "works", false)
.await
.expect("terminal input remains available");
manager
.resize(terminal_id, 100, 30)
.await
.expect("terminal resize remains available");
}
#[tokio::test]
async fn delayed_activity_from_a_replaced_stream_is_discarded() {
let (_state, manager, structured_id, old_events, _terminal_id) =
let (_state, manager, _panes, structured_id, old_events, _terminal_id) =
manager_with_structured_stream().await;
let replacement = Arc::new(tokio::sync::Mutex::new(stream(
manager
+33
View File
@@ -1311,6 +1311,39 @@ async fn attach_ws_streams_pty_bytes() {
assert!(saw_hello, "PTY did not echo through the attach WS");
}
#[tokio::test]
async fn pty_routes_reject_structured_sessions_as_not_terminal() {
let h = spawn("naps-laptop").await;
let fake = FakeAcp::new();
let id = structured_session(&h, &fake).await;
let response = reqwest::Client::new()
.post(url(h.addr, &format!("/api/sessions/{id}/input")))
.json(&maestro_protocol::SendInputRequest {
text: "ignored".into(),
submit: false,
})
.send()
.await
.unwrap();
assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
let error: ApiError = response.json().await.unwrap();
assert_eq!(error.error.code, "not_terminal");
assert_eq!(error.error.message, "session is not a terminal session");
let error = connect_async(ws_url(h.addr, &format!("/api/attach/{id}")))
.await
.expect_err("structured session must not upgrade to a PTY socket");
let tokio_tungstenite::tungstenite::Error::Http(response) = error else {
panic!("expected an HTTP handshake rejection, got {error}");
};
assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
let body = response.body().as_deref().expect("ApiError response body");
let error: ApiError = serde_json::from_slice(body).unwrap();
assert_eq!(error.error.code, "not_terminal");
assert_eq!(error.error.message, "session is not a terminal session");
}
#[tokio::test]
async fn attach_ws_for_unknown_session_returns_404() {
let h = spawn("naps-laptop").await;
+4 -2
View File
@@ -162,8 +162,10 @@ choices, not daemon ones.
**A client decides:**
- How each row kind is drawn — markdown for `AgentText`/`AgentThought`, a
diff view for `ToolDiff`, a tool card for `ToolCall`, a checklist for
`Plan`, and a floor presentation for `Notice`.
diff view for `ToolDiff`, a tool card for `ToolCall`, and a floor
presentation for `Notice`. `Plan` is the exception: both clients omit it
from the transcript entirely and render the current plan in a separate
plan rail (web) / status line (TUI) instead.
- How `responding` vs. `thinking` are distinguished visually, if at all.
- How the operator is asked to resolve an `Approval` row — the daemon only
exposes the three-way decision (`AllowOnce` / `AllowForSession` / `Deny`),
+235
View File
@@ -44,6 +44,22 @@ function row(id: string, kind: TranscriptRow["kind"], data: unknown): Transcript
return { id, kind, data };
}
function toolRow(id: string, text: string): TranscriptRow {
return row(id, "tool_call", {
title: `Read(${id})`,
status: "completed",
content: [{ type: "content", content: { type: "text", text } }],
});
}
function diffRow(id: string, newText: string): TranscriptRow {
return row(id, "tool_diff", {
title: `Edit(${id})`,
status: "completed",
content: [{ type: "diff", path: "file.ts", oldText: "before", newText }],
});
}
const idleState: ReducedState = { turn: { responding: false, thinking: false, active_tool_calls: [] } };
const allApprovalOptions = [
@@ -93,6 +109,121 @@ describe("structured connection: data flow", () => {
conn.dispose();
});
it("keeps snapshot plan rows out of the transcript while rendering the plan rail", () => {
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
ws.open();
ws.emit({
type: "transcript_snapshot",
snapshot: {
cursor: 3,
rows: [
row("user:1", "user", { content: { type: "text", text: "before plan" } }),
row("plan", "plan", { entries: [{ content: "ship it", status: "pending" }] }),
row("agent:1", "agent_text", { content: { type: "text", text: "after plan" } }),
],
},
});
ws.emit({
type: "reduced_state",
state: {
...idleState,
plan: { entries: [{ content: "ship it", priority: "high", status: "pending" }] },
},
});
const transcript = conn.el.querySelector(".structured-rows")!;
expect(transcript.querySelectorAll(".st-row")).toHaveLength(2);
expect(transcript.querySelector('[data-row-id="plan"]')).toBeNull();
expect(transcript.textContent).toContain("before plan");
expect(transcript.textContent).toContain("after plan");
expect(transcript.textContent).not.toContain("[plan]");
expect(conn.el.querySelector(".structured-rail")!.textContent).toContain("ship it");
conn.dispose();
});
it("applies plan append, patch, and remove without disturbing surrounding rows", () => {
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
ws.open();
ws.emit({
type: "transcript_snapshot",
snapshot: {
cursor: 1,
rows: [row("user:1", "user", { content: { type: "text", text: "keep me" } })],
},
});
ws.emit({ type: "reduced_state", state: idleState });
ws.emit({
type: "projection_update",
transcript: {
type: "append",
row: row("plan", "plan", { entries: [{ content: "first", status: "pending" }] }),
},
state: {
...idleState,
plan: { entries: [{ content: "first", priority: "high", status: "pending" }] },
},
});
ws.emit({
type: "projection_update",
transcript: {
type: "append",
row: row("agent:1", "agent_text", { content: { type: "text", text: "still here" } }),
},
state: {
...idleState,
plan: { entries: [{ content: "first", priority: "high", status: "pending" }] },
},
});
ws.emit({
type: "projection_update",
transcript: {
type: "patch",
row: row("plan", "plan", { entries: [{ content: "updated", status: "completed" }] }),
},
state: {
...idleState,
plan: { entries: [{ content: "updated", priority: "high", status: "completed" }] },
},
});
const transcript = conn.el.querySelector(".structured-rows")!;
expect([...transcript.querySelectorAll<HTMLElement>(".st-row")].map((el) => el.dataset.rowId)).toEqual([
"user:1",
"agent:1",
]);
expect(conn.el.querySelector(".structured-rail")!.textContent).toContain("updated");
expect(conn.el.querySelector(".structured-rail")!.textContent).not.toContain("first");
ws.emit({
type: "projection_update",
transcript: { type: "remove", id: "plan" },
state: idleState,
});
ws.emit({
type: "projection_update",
transcript: {
type: "patch",
row: row("agent:1", "agent_text", { content: { type: "text", text: "patched neighbor" } }),
},
state: idleState,
});
ws.emit({
type: "projection_update",
transcript: { type: "remove", id: "user:1" },
state: idleState,
});
expect(conn.el.querySelector(".structured-rail .st-plan")).toBeNull();
expect([...transcript.querySelectorAll<HTMLElement>(".st-row")].map((el) => el.dataset.rowId)).toEqual([
"agent:1",
]);
expect(transcript.textContent).toContain("patched neighbor");
conn.dispose();
});
it("appends and patches rows via projection_update without duplicating them", () => {
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
@@ -535,6 +666,110 @@ describe("structured connection: reconnection/error/loading", () => {
conn.dispose();
});
it("does not mount plan rows from a reconnect snapshot", () => {
const first = new FakeSocket();
const second = new FakeSocket();
const conn = mount({ sockets: [first, second], shouldReconnect: () => true });
first.open();
first.emit({
type: "transcript_snapshot",
snapshot: {
cursor: 1,
rows: [row("agent:old", "agent_text", { content: { type: "text", text: "old" } })],
},
});
first.drop();
vi.advanceTimersByTime(1000);
second.open();
second.emit({
type: "transcript_snapshot",
snapshot: {
cursor: 4,
rows: [
row("user:new", "user", { content: { type: "text", text: "new" } }),
row("plan", "plan", { entries: [{ content: "reconnected", status: "in_progress" }] }),
],
},
});
second.emit({
type: "reduced_state",
state: {
...idleState,
plan: { entries: [{ content: "reconnected", priority: "high", status: "in_progress" }] },
},
});
const transcript = conn.el.querySelector(".structured-rows")!;
expect(transcript.textContent).toBe("Younew");
expect(transcript.querySelector('[data-row-id="plan"]')).toBeNull();
expect(conn.el.querySelector(".structured-rail")!.textContent).toContain("reconnected");
conn.dispose();
});
it("keeps surviving expanded rows open across a reconnect snapshot and leaves fresh rows collapsed", () => {
const first = new FakeSocket();
const second = new FakeSocket();
const conn = mount({ sockets: [first, second], shouldReconnect: () => true });
first.open();
first.emit({
type: "transcript_snapshot",
snapshot: { cursor: 2, rows: [toolRow("tool:1", "before"), diffRow("diff:1", "first")] },
});
for (const head of conn.el.querySelectorAll<HTMLButtonElement>(".st-tool-head")) head.click();
first.drop();
vi.advanceTimersByTime(1000);
second.open();
second.emit({
type: "transcript_snapshot",
snapshot: {
cursor: 3,
rows: [toolRow("tool:1", "after"), diffRow("diff:1", "second"), toolRow("tool:2", "new")],
},
});
expect(
conn.el.querySelector('[data-row-id="tool:1"] .st-tool-head')?.getAttribute("aria-expanded"),
).toBe("true");
expect(
conn.el.querySelector('[data-row-id="diff:1"] .st-tool-head')?.getAttribute("aria-expanded"),
).toBe("true");
expect(
conn.el.querySelector('[data-row-id="tool:2"] .st-tool-head')?.getAttribute("aria-expanded"),
).toBe("false");
expect(conn.el.querySelector('[data-row-id="tool:1"] .st-tool-out')?.textContent).toContain("after");
conn.dispose();
});
it("prunes expanded IDs removed or made non-expandable by a snapshot", () => {
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
ws.open();
ws.emit({
type: "transcript_snapshot",
snapshot: { cursor: 2, rows: [toolRow("tool:gone", "gone"), toolRow("tool:plain", "plain")] },
});
for (const head of conn.el.querySelectorAll<HTMLButtonElement>(".st-tool-head")) head.click();
ws.emit({
type: "transcript_snapshot",
snapshot: {
cursor: 3,
rows: [row("tool:plain", "agent_text", { content: { type: "text", text: "not a tool" } })],
},
});
ws.emit({
type: "transcript_snapshot",
snapshot: { cursor: 5, rows: [toolRow("tool:gone", "returned"), toolRow("tool:plain", "returned")] },
});
for (const head of conn.el.querySelectorAll<HTMLButtonElement>(".st-tool-head")) {
expect(head.getAttribute("aria-expanded")).toBe("false");
}
conn.dispose();
});
it("keeps a local submission gated across a reconnect until begin-turn arrives", () => {
vi.mocked(sendStructuredPrompt).mockImplementationOnce(() => new Promise(() => {}));
const first = new FakeSocket();
+50 -8
View File
@@ -106,6 +106,7 @@ export function createStructuredConnection(
let rows: TranscriptRow[] = [];
const rowIndex = new Map<string, number>();
const rowEls = new Map<string, HTMLElement>();
const expandedRowIds = new Set<string>();
let state: ReducedState = { turn: idleTurn() };
let pendingApprovalRowId: string | null = null;
let latestSession: Session | undefined = store?.snapshot().sessions.get(sessionId);
@@ -151,27 +152,66 @@ export function createStructuredConnection(
}
}
function mountRow(row: TranscriptRow): void {
const el = buildTranscriptRow(row, approvalStateFor(row));
function buildRowElement(row: TranscriptRow): HTMLElement | null {
const el = buildTranscriptRow(
row,
approvalStateFor(row),
expandedRowIds.has(row.id),
(expanded) => {
if (expanded) expandedRowIds.add(row.id);
else expandedRowIds.delete(row.id);
},
);
if (!el) return null;
if (el.querySelector(".st-tool-head")?.getAttribute("aria-expanded") !== "true") {
expandedRowIds.delete(row.id);
}
return el;
}
function insertRowElement(row: TranscriptRow, el: HTMLElement): void {
const index = rowIndex.get(row.id) ?? rows.length - 1;
let next: HTMLElement | null = null;
for (let i = index + 1; i < rows.length; i++) {
next = rowEls.get(rows[i].id) ?? null;
if (next) break;
}
rowEls.set(row.id, el);
rowsEl.appendChild(el);
rowsEl.insertBefore(el, next);
}
function mountRow(row: TranscriptRow): void {
const el = buildRowElement(row);
if (el) insertRowElement(row, el);
}
function remountRow(row: TranscriptRow): void {
const old = rowEls.get(row.id);
const oldToolHead = old?.querySelector<HTMLButtonElement>(".st-tool-head") ?? null;
const wasExpanded = oldToolHead?.getAttribute("aria-expanded") === "true";
const restoreFocus = document.activeElement === oldToolHead;
const el = buildTranscriptRow(row, approvalStateFor(row), wasExpanded);
rowEls.set(row.id, el);
if (old) old.replaceWith(el);
else rowsEl.appendChild(el);
const el = buildRowElement(row);
if (!el) {
old?.remove();
rowEls.delete(row.id);
expandedRowIds.delete(row.id);
return;
}
if (old) {
rowEls.set(row.id, el);
old.replaceWith(el);
} else {
insertRowElement(row, el);
}
const newToolHead = el.querySelector<HTMLButtonElement>(".st-tool-head");
if (restoreFocus) newToolHead?.focus();
}
function applySnapshot(snapshot: { rows: TranscriptRow[] }): void {
rows = [...snapshot.rows];
const expandableIds = new Set(
rows.filter((row) => row.kind === "tool_call" || row.kind === "tool_diff").map((row) => row.id),
);
for (const id of expandedRowIds) if (!expandableIds.has(id)) expandedRowIds.delete(id);
rowIndex.clear();
rows.forEach((row, i) => rowIndex.set(row.id, i));
rowEls.clear();
@@ -199,6 +239,7 @@ export function createStructuredConnection(
const i = rowIndex.get(delta.id);
if (i === undefined) return;
rows.splice(i, 1);
expandedRowIds.delete(delta.id);
rowIndex.clear();
rows.forEach((row, idx) => rowIndex.set(row.id, idx));
rowEls.get(delta.id)?.remove();
@@ -482,6 +523,7 @@ export function createStructuredConnection(
resizeObserver.disconnect();
unsubscribeStore?.();
closeCurrentSocket();
expandedRowIds.clear();
el.remove();
},
};
+15 -4
View File
@@ -9,7 +9,11 @@ import {
} from "./structuredRender.js";
import type { Session, TranscriptRow } from "./types.js";
function row(kind: TranscriptRow["kind"], data: unknown, id = "r1"): TranscriptRow {
function row<K extends TranscriptRow["kind"]>(
kind: K,
data: unknown,
id = "r1",
): TranscriptRow & { kind: K } {
return { id, kind, data };
}
@@ -275,9 +279,16 @@ describe("buildTranscriptRow: approval", () => {
});
describe("buildTranscriptRow: notice / unknown kinds", () => {
it("renders a plan or notice row without crashing", () => {
expect(() => buildTranscriptRow(row("notice", { sessionUpdate: "future" }))).not.toThrow();
expect(() => buildTranscriptRow(row("plan", { entries: [] }))).not.toThrow();
it("renders notice and unknown row kinds safely", () => {
expect(buildTranscriptRow(row("notice", { sessionUpdate: "future" }))?.textContent).toBe(
"[notice]",
);
const unknown = { id: "r1", kind: "future_kind", data: {} } as unknown as TranscriptRow;
expect(buildTranscriptRow(unknown)?.textContent).toBe("[future_kind]");
});
it("returns no transcript element for a plan row", () => {
expect(buildTranscriptRow(row("plan", { entries: [] }))).toBeNull();
});
});
+32 -5
View File
@@ -187,7 +187,11 @@ function statusMeta(status: string | undefined, hasDiff: boolean): { text: strin
return null;
}
function buildToolRow(row: TranscriptRow, expanded: boolean): HTMLElement {
function buildToolRow(
row: TranscriptRow,
expanded: boolean,
onExpandedChange?: (expanded: boolean) => void,
): HTMLElement {
const tool = (row.data ?? {}) as AcpToolCallLike;
const content = tool.content ?? [];
const hasDiff = content.some((c) => c.type === "diff");
@@ -259,6 +263,7 @@ function buildToolRow(row: TranscriptRow, expanded: boolean): HTMLElement {
};
head.addEventListener("click", () => {
setExpanded(head.getAttribute("aria-expanded") !== "true");
onExpandedChange?.(head.getAttribute("aria-expanded") === "true");
});
setExpanded(expanded);
@@ -362,13 +367,33 @@ function buildApprovalRow(row: TranscriptRow, approval: ApprovalRowState): HTMLE
return wrap;
}
/** Build the DOM for one transcript row. `approval` is consulted only for an
* `approval` row; pass it whenever the row might be that kind. */
/** Build the DOM for one transcript row, or `null` for a plan shown only in
* the rail. `approval` is consulted only for an `approval` row; tool
* expansion is supplied and reported explicitly. */
export function buildTranscriptRow(
row: TranscriptRow & { kind: "plan" },
approval?: ApprovalRowState,
toolExpanded?: boolean,
onToolExpandedChange?: (expanded: boolean) => void,
): null;
export function buildTranscriptRow(
row: TranscriptRow & { kind: Exclude<TranscriptRow["kind"], "plan"> },
approval?: ApprovalRowState,
toolExpanded?: boolean,
onToolExpandedChange?: (expanded: boolean) => void,
): HTMLElement;
export function buildTranscriptRow(
row: TranscriptRow,
approval?: ApprovalRowState,
toolExpanded?: boolean,
onToolExpandedChange?: (expanded: boolean) => void,
): HTMLElement | null;
export function buildTranscriptRow(
row: TranscriptRow,
approval?: ApprovalRowState,
toolExpanded = false,
): HTMLElement {
onToolExpandedChange?: (expanded: boolean) => void,
): HTMLElement | null {
let built: HTMLElement;
switch (row.kind) {
case "user":
@@ -382,11 +407,13 @@ export function buildTranscriptRow(
break;
case "tool_call":
case "tool_diff":
built = buildToolRow(row, toolExpanded);
built = buildToolRow(row, toolExpanded, onToolExpandedChange);
break;
case "approval":
built = buildApprovalRow(row, approval ?? { pending: false, onDecide: () => Promise.resolve() });
break;
case "plan":
return null;
default:
built = buildPlainRow(row);
break;