Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b680266990 | ||
|
|
ded1f050b2 | ||
|
|
a58ebfed93 | ||
|
|
31b429a840 | ||
|
|
12bd68a2af | ||
|
|
f765d415d6 | ||
|
|
d3e3010245 |
@@ -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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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`),
|
||||
|
||||
@@ -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
@@ -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();
|
||||
},
|
||||
};
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user