naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(maestro): render structured TUI sessions

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 07:19:10 +01:00
parent 8c97ef0616
commit e9ede49256
13 changed files with 2025 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -1644,6 +1644,7 @@ dependencies = [
"tokio",
"tokio-tungstenite 0.21.0",
"tracing",
"unicode-width",
"uuid",
"vt100",
]
Cargo.toml
+1
View File
@@ -44,6 +44,7 @@ tokio-util = { version = "0.7", features = ["compat"] }
toml = "0.8"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
unicode-width = "0.1"
uuid = { version = "1", features = ["v4", "serde"] }
ratatui = "0.28"
vt100 = "0.15"
crates/maestro-protocol/src/client.rs
+57 -1
View File
@@ -17,7 +17,7 @@ use crate::routes::{self, UrlError};
use crate::{
ApiError, BranchInfo, CreateSessionRequest, DaemonInfo, DirListing, ErrorInfo, GitRoot,
Project, RecentDirs, RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session,
SetFocusRequest,
SetFocusRequest, StructuredApprovalDecision, StructuredPromptRequest,
};
/// Time budget for establishing a TCP connection to the daemon.
@@ -105,6 +105,11 @@ impl DaemonClient {
Ok(routes::attach_url(&self.base, id)?)
}
/// `WS /api/structured/:id` URL for this daemon.
pub fn structured_url(&self, id: Uuid) -> Result<String, ClientError> {
Ok(routes::structured_url(&self.base, id)?)
}
// -- typed calls --------------------------------------------------------
pub async fn list_sessions(&self) -> Result<Vec<Session>, ClientError> {
@@ -179,6 +184,57 @@ impl DaemonClient {
Ok(())
}
/// Send one prompt to a structured session. The daemon rejects a second
/// prompt while the preceding ACP turn is still active.
pub async fn send_structured_prompt(&self, id: Uuid, prompt: &str) -> Result<(), ClientError> {
let req = StructuredPromptRequest {
prompt: prompt.to_string(),
};
let resp = self
.send(
Method::POST,
&routes::structured_prompt(id),
&[],
Some(&req),
)
.await?;
check(resp).await?;
Ok(())
}
/// Cancel the structured session's active ACP turn.
pub async fn cancel_structured_turn(&self, id: Uuid) -> Result<(), ClientError> {
let resp = self
.send(
Method::POST,
&routes::structured_cancel(id),
&[],
None::<&()>,
)
.await?;
check(resp).await?;
Ok(())
}
/// Resolve the currently pending structured-session permission request.
pub async fn resolve_structured_approval(
&self,
id: Uuid,
nonce: &str,
decision: StructuredApprovalDecision,
) -> Result<(), ClientError> {
let resp = self
.send(
Method::POST,
&routes::structured_approval(id, nonce),
&[],
Some(&decision),
)
.await?;
check(resp).await?;
Ok(())
}
pub async fn daemon_info(&self) -> Result<DaemonInfo, ClientError> {
self.get(routes::INFO, &[]).await
}
crates/maestro-protocol/src/routes.rs
+9
View File
@@ -159,6 +159,11 @@ pub fn attach_url(http_base: &str, id: Uuid) -> Result<String, UrlError> {
Ok(format!("{}{}", ws_base(http_base)?, attach(id)))
}
/// Full `WS /api/structured/:id` URL for a daemon reachable at `http_base`.
pub fn structured_url(http_base: &str, id: Uuid) -> Result<String, UrlError> {
Ok(format!("{}{}", ws_base(http_base)?, structured(id)))
}
#[cfg(test)]
mod tests {
use super::*;
@@ -208,6 +213,10 @@ mod tests {
events_url("http://host:7070/").unwrap(),
"ws://host:7070/api/events"
);
assert_eq!(
structured_url("http://host:7070/", id()).unwrap(),
"ws://host:7070/api/structured/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
);
}
#[test]
crates/maestro-tui/Cargo.toml
+1
View File
@@ -21,6 +21,7 @@ thiserror.workspace = true
tokio = { workspace = true, features = ["sync", "time", "macros", "rt"] }
tokio-tungstenite.workspace = true
tracing.workspace = true
unicode-width.workspace = true
uuid.workspace = true
vt100.workspace = true
crates/maestro-tui/src/app.rs
+336 -12
View File
@@ -17,7 +17,10 @@ use std::time::Duration;
use chrono::Utc;
use crossterm::event::{Event as CtEvent, EventStream};
use futures_util::StreamExt;
use maestro_protocol::{routes, BranchInfo, CreateSessionRequest, DaemonClient, NameBatch};
use maestro_protocol::{
routes, BranchInfo, CreateSessionRequest, DaemonClient, NameBatch, SessionView,
StructuredApprovalDecision,
};
use ratatui::Terminal;
use uuid::Uuid;
@@ -38,6 +41,8 @@ use crate::render::{draw, RegionMap, RenderInputs};
use crate::session_task::{SessionInput, SessionTask};
use crate::sidebar::Sidebar;
use crate::state::{AppState, Focus, OverlayKind, SidebarItem};
use crate::structured_pane::StructuredPane;
use crate::structured_task::{StructuredHandle, StructuredTask};
use crate::text_input::TextInput;
use crate::TuiError;
@@ -54,6 +59,7 @@ pub struct App {
home_dir: String,
catalog: CatalogStore,
emulators: SessionEmulators,
structured: HashMap<Uuid, StructuredHandle>,
state: AppState,
help: HelpOverlay,
chord: ChordState,
@@ -163,6 +169,18 @@ pub enum Effect {
id: Uuid,
bytes: Vec<u8>,
},
StructuredPrompt {
id: Uuid,
prompt: String,
},
StructuredCancel {
id: Uuid,
},
StructuredApproval {
id: Uuid,
nonce: String,
decision: StructuredApprovalDecision,
},
/// Fetch presets / projects / names for a freshly opened form (#64).
LoadDraft,
/// Fast-forward spawn (#67): resolve every default from the daemon and
@@ -214,6 +232,7 @@ impl App {
home_dir: std::env::var("HOME").unwrap_or_default(),
catalog,
emulators: SessionEmulators::default(),
structured: HashMap::new(),
state: AppState::new(),
draft: None,
help,
@@ -321,6 +340,11 @@ impl App {
return Ok(false);
}
if let Some(effects) = self.handle_structured_key(&key) {
self.perform(effects).await;
return Ok(false);
}
if let Some(effect) = self.handle_pane_scroll(&key) {
self.perform(effect.into_iter().collect()).await;
return Ok(false);
@@ -419,6 +443,44 @@ impl App {
let _ = handle.tx.send(SessionInput::Bytes(bytes)).await;
}
}
Effect::StructuredPrompt { id, prompt } => {
if let Err(error) = self.daemon.send_structured_prompt(id, &prompt).await {
if let Some(handle) = self.structured.get(&id) {
handle
.state
.lock()
.expect("structured state mutex")
.prompt_failed(prompt, error.to_string());
}
}
}
Effect::StructuredCancel { id } => {
if let Err(error) = self.daemon.cancel_structured_turn(id).await {
if let Some(handle) = self.structured.get(&id) {
handle.state.lock().expect("structured state mutex").error =
Some(error.to_string());
}
}
}
Effect::StructuredApproval {
id,
nonce,
decision,
} => {
if let Err(error) = self
.daemon
.resolve_structured_approval(id, &nonce, decision)
.await
{
if let Some(handle) = self.structured.get(&id) {
handle
.state
.lock()
.expect("structured state mutex")
.approval_failed(error.to_string());
}
}
}
}
}
}
@@ -563,13 +625,13 @@ impl App {
Action::OpenCommandBar => effects.extend(self.open_command_bar()),
// Escape hatch (#156): the literal byte, not the bar.
Action::SendLiteralCtrlK => {
effects.extend(self.forward_to_session(&InputKey::ctrl_char('k')));
effects.extend(self.forward_literal_to_terminal(&InputKey::ctrl_char('k')));
}
// Escape hatch (#158): the leader's own byte, built from the
// configured leader rather than a hardcoded key.
Action::SendLiteralLeader => {
let leader = self.cfg.leader.clone();
effects.extend(self.forward_to_session(&leader));
effects.extend(self.forward_literal_to_terminal(&leader));
}
Action::OpenSpawnModal => {
self.open_spawn_modal();
@@ -587,7 +649,7 @@ impl App {
}
// A previewed session is already attached: committing
// to it keeps that stream rather than reopening one.
if self.emulators.get(sid).is_none() {
if !self.is_attached(sid) {
effects.push(Effect::Attach(sid));
}
}
@@ -595,6 +657,9 @@ impl App {
// preview budget.
self.preview_lru.retain(|s| *s != sid);
self.state.attach_to_focused(sid);
if self.session_view(sid) == SessionView::Structured {
self.state.mode = Mode::Normal;
}
}
}
Action::KillHighlighted => {
@@ -663,7 +728,7 @@ impl App {
self.state.preview = Some(sid);
// Set outside the input path, so nothing else marks the frame stale.
self.epoch = self.epoch.wrapping_add(1);
let fresh = self.emulators.get(sid).is_none();
let fresh = !self.is_attached(sid);
if fresh {
self.preview_lru.retain(|s| *s != sid);
self.preview_lru.push(sid);
@@ -732,15 +797,23 @@ impl App {
/// Open a stream for `sid` if one isn't already running.
fn attach(&mut self, sid: Uuid) {
if self.emulators.get(sid).is_some() {
if self.is_attached(sid) {
return;
}
match self.daemon.attach_url(sid) {
Ok(ws_url) => {
let handle = SessionTask::spawn(ws_url, 24, 80);
self.emulators.insert(sid, handle);
}
Err(e) => self.state.toast(format!("attach failed: {e}")),
match self.session_view(sid) {
SessionView::Terminal => match self.daemon.attach_url(sid) {
Ok(ws_url) => {
let handle = SessionTask::spawn(ws_url, 24, 80);
self.emulators.insert(sid, handle);
}
Err(e) => self.state.toast(format!("attach failed: {e}")),
},
SessionView::Structured => match self.daemon.structured_url(sid) {
Ok(ws_url) => {
self.structured.insert(sid, StructuredTask::spawn(ws_url));
}
Err(e) => self.state.toast(format!("attach failed: {e}")),
},
}
}
@@ -748,9 +821,26 @@ impl App {
/// dropping its vt100 parser.
fn release_session(&mut self, sid: Uuid) {
self.emulators.remove(sid);
if let Some(handle) = self.structured.remove(&sid) {
handle.end();
}
self.last_sent_size.remove(&sid);
}
fn session_view(&self, sid: Uuid) -> SessionView {
self.catalog
.snapshot()
.sessions
.iter()
.find(|session| session.id == sid)
.map(|session| session.view)
.unwrap_or_default()
}
fn is_attached(&self, sid: Uuid) -> bool {
self.emulators.get(sid).is_some() || self.structured.contains_key(&sid)
}
/// Return the session under the sidebar highlight, falling back to the
/// first listed session so the user can attach without arrowing first.
fn pick_sidebar_target(&self) -> Option<Uuid> {
@@ -817,6 +907,16 @@ impl App {
/// attached to send it to.
fn forward_to_session(&mut self, key: &InputKey) -> Option<Effect> {
let sid = self.state.active_session()?;
if let Some(handle) = self.structured.get(&sid) {
let mut state = handle.state.lock().expect("structured state mutex");
if matches!(key, InputKey::Enter) {
return state
.begin_prompt()
.map(|prompt| Effect::StructuredPrompt { id: sid, prompt });
}
state.edit_composer(key);
return None;
}
let handle = self.emulators.get(sid)?;
let bytes = encode_key(key)?;
handle
@@ -827,6 +927,86 @@ impl App {
Some(Effect::Forward { id: sid, bytes })
}
fn forward_literal_to_terminal(&mut self, key: &InputKey) -> Option<Effect> {
let sid = self.state.active_session()?;
if self.session_view(sid) == SessionView::Structured {
return None;
}
self.forward_to_session(key)
}
fn handle_structured_key(&mut self, key: &InputKey) -> Option<Vec<Effect>> {
if self.state.focus != Focus::Pane || self.state.preview.is_some() {
return None;
}
let sid = self.state.active_session()?;
let handle = self.structured.get(&sid)?;
let mut state = handle.state.lock().expect("structured state mutex");
if self.state.mode == Mode::Insert {
if matches!(key, InputKey::Esc) && state.turn_active() {
return Some(vec![Effect::StructuredCancel { id: sid }]);
}
return None;
}
let effects = match key {
InputKey::Char {
c: 'j',
ctrl: false,
alt: false,
..
}
| InputKey::Down => {
state.select(1);
Vec::new()
}
InputKey::Char {
c: 'k',
ctrl: false,
alt: false,
..
}
| InputKey::Up => {
state.select(-1);
Vec::new()
}
InputKey::Enter => {
state.toggle_selected();
Vec::new()
}
InputKey::Esc => {
if state.turn_active() {
vec![Effect::StructuredCancel { id: sid }]
} else {
Vec::new()
}
}
InputKey::Char {
c: key @ ('a' | 'A' | 'd'),
ctrl: false,
alt: false,
..
} => state
.approval_decision(*key)
.map(|(nonce, decision)| Effect::StructuredApproval {
id: sid,
nonce,
decision,
})
.into_iter()
.collect(),
InputKey::PageUp => {
state.scroll_by(self.pane_page_rows() as isize);
Vec::new()
}
InputKey::PageDown => {
state.scroll_by(-(self.pane_page_rows() as isize));
Vec::new()
}
_ => return None,
};
Some(effects)
}
fn pane_page_rows(&self) -> usize {
self.region
.pane
@@ -839,6 +1019,14 @@ impl App {
let Some(sid) = self.state.active_session() else {
return;
};
if let Some(handle) = self.structured.get(&sid) {
handle
.state
.lock()
.expect("structured state mutex")
.scroll_by(rows);
return;
}
let Some(handle) = self.emulators.get(sid) else {
return;
};
@@ -872,6 +1060,14 @@ impl App {
_ => return None,
};
let sid = self.state.active_session()?;
if let Some(handle) = self.structured.get(&sid) {
handle
.state
.lock()
.expect("structured state mutex")
.scroll_by(lines);
return Some(None);
}
let handle = self.emulators.get(sid)?;
let pane = self.region.pane?;
let parser = handle.screen.lock().expect("vt100 parser mutex");
@@ -1294,6 +1490,9 @@ impl App {
}
self.draft = None;
self.state.complete_spawn(sid);
if session.view == SessionView::Structured {
self.state.mode = Mode::Normal;
}
self.attach(sid);
}
Err(e) => {
@@ -1355,6 +1554,11 @@ impl App {
self.emulators
.get(sid)
.map(|h| (sid, h.revision(), h.state()))
.or_else(|| {
self.structured
.get(&sid)
.map(|h| (sid, h.revision(), h.connection()))
})
});
FrameKey {
catalog: self.catalog.version(),
@@ -1416,6 +1620,17 @@ impl App {
focused: matches!(self.state.focus, Focus::Pane) && !previewing,
})
});
let structured_pane = shown_sid.and_then(|sid| {
let handle = self.structured.get(&sid)?;
let session = snap.sessions.iter().find(|session| session.id == sid)?;
Some(StructuredPane {
handle: handle.clone(),
focused: matches!(self.state.focus, Focus::Pane) && !previewing,
mode: self.state.mode,
session_status: session.status,
activity_detail: session.activity_detail.clone(),
})
});
let pane_title = match shown_sid {
Some(sid) => {
let name = snap
@@ -1439,6 +1654,7 @@ impl App {
home_dir: &self.home_dir,
pane_title: &pane_title,
pane: pane.as_ref(),
structured_pane: structured_pane.as_ref(),
state: &self.state,
draft: self.draft.as_ref(),
help: &self.help,
@@ -3697,4 +3913,112 @@ mod tests {
assert!(app.command_bar.is_none());
assert!(app.state.overlay().is_none());
}
fn structured_app(sid: Uuid) -> App {
let mut app = app_with_highlighted_session(sid);
let mut session = app.catalog.snapshot().sessions[0].clone();
session.view = SessionView::Structured;
app.catalog
.apply_event(maestro_protocol::CatalogEvent::SessionUpdated { session });
app.structured.insert(
sid,
StructuredHandle::test_handle(crate::structured_task::StructuredState::default()),
);
app.state.attach_to_focused(sid);
app
}
#[tokio::test]
async fn session_view_selects_the_structured_connection_and_cleanup_stops_it() {
let sid = Uuid::from_u128(141);
let mut app = app_with_highlighted_session(sid);
let mut session = app.catalog.snapshot().sessions[0].clone();
session.view = SessionView::Structured;
app.catalog
.apply_event(maestro_protocol::CatalogEvent::SessionUpdated { session });
app.attach(sid);
assert!(app.structured.contains_key(&sid));
assert!(app.emulators.get(sid).is_none());
let handle = app.structured.get(&sid).unwrap().clone();
app.release_session(sid);
assert_eq!(
handle.connection(),
crate::session_task::SessionState::Ended
);
}
#[test]
fn structured_normal_mode_routes_navigation_expansion_and_approval_by_row() {
use maestro_protocol::{ReducedState, TranscriptRow, TranscriptRowKind};
use serde_json::json;
let sid = Uuid::from_u128(141);
let mut app = structured_app(sid);
app.dispatch(Action::AttachHighlighted);
assert_eq!(app.state.focus, Focus::Pane);
assert_eq!(app.state.mode, Mode::Normal);
let handle = app.structured.get(&sid).unwrap().clone();
{
let mut state = handle.state.lock().unwrap();
state.rows = vec![
TranscriptRow {
id: "text".into(),
kind: TranscriptRowKind::AgentText,
data: json!({"content":{"type":"text","text":"hello"}}),
},
TranscriptRow {
id: "approval".into(),
kind: TranscriptRowKind::Approval,
data: json!({"options":[{"kind":"allow_once"}]}),
},
];
state.selected = Some("text".into());
state.reduced = ReducedState {
pending_approval: Some(json!({"nonce":"n"})),
..ReducedState::default()
};
}
assert_eq!(
app.handle_structured_key(&InputKey::char('j')),
Some(vec![])
);
assert_eq!(
app.handle_structured_key(&InputKey::char('a')),
Some(vec![Effect::StructuredApproval {
id: sid,
nonce: "n".into(),
decision: StructuredApprovalDecision::AllowOnce,
}])
);
assert_eq!(
handle.state.lock().unwrap().selected.as_deref(),
Some("approval")
);
}
#[test]
fn structured_insert_mode_edits_submits_and_cancels_without_pty_bytes() {
let sid = Uuid::from_u128(141);
let mut app = structured_app(sid);
assert!(app.emulators.get(sid).is_none());
app.forward_to_session(&InputKey::char('h'));
app.forward_to_session(&InputKey::char('i'));
assert_eq!(
app.forward_to_session(&InputKey::Enter),
Some(Effect::StructuredPrompt {
id: sid,
prompt: "hi".into(),
})
);
assert_eq!(
app.handle_structured_key(&InputKey::Esc),
Some(vec![Effect::StructuredCancel { id: sid }])
);
assert_eq!(
app.forward_literal_to_terminal(&app.cfg.leader.clone()),
None
);
}
}
crates/maestro-tui/src/lib.rs
+2
View File
@@ -23,6 +23,8 @@ pub mod render;
pub mod session_task;
pub mod sidebar;
pub mod state;
pub mod structured_pane;
pub mod structured_task;
pub mod terminal;
pub mod text_input;
crates/maestro-tui/src/render.rs
+30 -6
View File
@@ -18,6 +18,7 @@ use crate::palette;
use crate::pane::Pane;
use crate::sidebar::Sidebar;
use crate::state::{AppState, Focus, OverlayKind, SidebarItem};
use crate::structured_pane::StructuredPane;
/// Tagged regions for mouse hit-testing. The input router resolves a mouse
/// coordinate to one of these via [`RegionMap::region_at`].
@@ -80,6 +81,7 @@ pub struct RenderInputs<'a> {
pub home_dir: &'a str,
pub pane_title: &'a str,
pub pane: Option<&'a Pane>,
pub structured_pane: Option<&'a StructuredPane>,
/// The spawn form behind [`OverlayKind::Spawn`], when one is up.
pub draft: Option<&'a crate::draft::DraftPane>,
pub state: &'a AppState,
@@ -147,14 +149,22 @@ pub fn draw(f: &mut Frame, inputs: RenderInputs<'_>) -> RegionMap {
render_title_bar(f, title_area, inputs.pane_title, pane_focused);
if let Some(p) = inputs.pane {
p.render(f, pane_area);
} else if let Some(p) = inputs.structured_pane {
p.render(f, pane_area);
}
// `RegionMap::pane` is also the PTY resize source. The title bar is
// chrome; every remaining cell belongs to the emulator.
// chrome; every remaining cell belongs to the active session pane.
map.pane = Some(pane_area);
// Status bar — always visible, never eclipsed by overlays so the user can
// always see the current mode + key hints.
render_status_bar(f, status_area, inputs.state, inputs.link);
render_status_bar(
f,
status_area,
inputs.state,
inputs.link,
inputs.structured_pane.is_some(),
);
match inputs.state.overlay() {
Some(OverlayKind::Spawn) => {
@@ -253,7 +263,13 @@ fn render_title_bar(f: &mut Frame, area: Rect, title: &str, focused: bool) {
/// Render the always-visible status bar at the bottom of the frame. Surfaces
/// mode + focused panel + the most useful key hints so the user is never
/// guessing what state they're in.
fn render_status_bar(f: &mut Frame, area: Rect, state: &AppState, link: &LinkStatus) {
fn render_status_bar(
f: &mut Frame,
area: Rect,
state: &AppState,
link: &LinkStatus,
structured: bool,
) {
// The daemon indicator is pinned right and sized to its own text, so a
// narrow terminal truncates the key hints rather than the release.
let label = link_label(link);
@@ -271,9 +287,13 @@ fn render_status_bar(f: &mut Frame, area: Rect, state: &AppState, link: &LinkSta
Focus::Sidebar => "sidebar",
Focus::Pane => "pane",
};
let hint = match state.mode {
Mode::Normal => "j/k: nav • enter: attach • s: spawn • ?: help • ^Wq: quit",
Mode::Insert => "^a: return to sidebar",
let hint = match (structured && state.focus == Focus::Pane, state.mode) {
(true, Mode::Normal) => "j/k: rows • enter: expand • i: compose • esc: cancel",
(true, Mode::Insert) => "enter: send • esc: cancel • ^a: sidebar",
(false, Mode::Normal) => {
"j/k: nav • enter: attach • s: spawn • ?: help • ^Wq: quit"
}
(false, Mode::Insert) => "^a: return to sidebar",
};
let mode_style = palette::active().mode_style(state.mode == Mode::Insert);
let line = ratatui::text::Line::from(vec![
@@ -435,6 +455,7 @@ mod tests {
home_dir: "",
pane_title: "session",
pane: None,
structured_pane: None,
draft: None,
state: &state,
help: &help,
@@ -532,6 +553,7 @@ mod tests {
home_dir: "",
pane_title: "session",
pane: None,
structured_pane: None,
draft: Some(&draft),
state: &state,
help: &help,
@@ -587,6 +609,7 @@ mod tests {
home_dir: "",
pane_title: "session",
pane: None,
structured_pane: None,
draft: None,
state: &state,
help: &help,
@@ -629,6 +652,7 @@ mod tests {
home_dir: "",
pane_title: "session",
pane: None,
structured_pane: None,
draft: None,
state: &state,
help: &help,
crates/maestro-tui/src/session_task.rs
+1 -1
View File
@@ -221,7 +221,7 @@ impl SessionTask {
}
}
fn jitter(base: Duration) -> Duration {
pub(crate) fn jitter(base: Duration) -> Duration {
let ms = base.as_millis() as u64;
let now_nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
crates/maestro-tui/src/structured_pane.rs
+801
View File
@@ -0,0 +1,801 @@
//! Native ratatui renderer for daemon-projected structured session rows.
use std::time::Duration;
use maestro_protocol::{SessionStatus, TranscriptRow, TranscriptRowKind};
use ratatui::layout::{Constraint, Direction, Layout, Rect};
use ratatui::style::{Modifier, Style};
use ratatui::text::{Line, Span};
use ratatui::widgets::Paragraph;
use ratatui::Frame;
use unicode_width::UnicodeWidthChar;
use crate::keymap::Mode;
use crate::palette;
use crate::session_task::SessionState;
use crate::structured_task::{offered_decisions, StructuredHandle, StructuredState};
const DIFF_PREVIEW_LINES: usize = 3;
#[derive(Debug, Clone)]
pub struct StructuredPane {
pub handle: StructuredHandle,
pub focused: bool,
pub mode: Mode,
pub session_status: SessionStatus,
pub activity_detail: Option<String>,
}
#[derive(Debug)]
struct VisualLine {
text: String,
row_id: Option<String>,
rail: bool,
tone: Tone,
}
#[derive(Debug, Clone, Copy)]
enum Tone {
Normal,
Muted,
Active,
Warn,
Error,
}
impl StructuredPane {
pub fn render(&self, f: &mut Frame, area: Rect) {
if area.width == 0 || area.height == 0 {
return;
}
let footer_height = area.height.min(3);
let parts = Layout::default()
.direction(Direction::Vertical)
.constraints([
Constraint::Min(0),
Constraint::Length(footer_height.saturating_sub(1).min(2)),
Constraint::Length(footer_height.min(1)),
])
.split(area);
let mut state = self.handle.state.lock().expect("structured state mutex");
self.render_transcript(f, parts[0], &mut state);
self.render_projection_status(f, parts[1], &state);
self.render_composer(f, parts[2], &state);
}
fn render_transcript(&self, f: &mut Frame, area: Rect, state: &mut StructuredState) {
let mut lines = build_lines(state, area.width.max(1) as usize);
if self.session_status == SessionStatus::Dead {
let detail = self
.activity_detail
.as_deref()
.unwrap_or("structured session ended");
lines.push(VisualLine {
text: format!("error: {detail}"),
row_id: None,
rail: true,
tone: Tone::Error,
});
} else {
match self.handle.connection() {
SessionState::Connecting => lines.push(system_line("connecting…")),
SessionState::Reconnecting => lines.push(system_line("reconnecting…")),
SessionState::Ended => lines.push(error_line("structured connection ended")),
SessionState::Live => {}
}
}
if let Some(error) = &state.error {
lines.push(error_line(error));
}
let viewport = usize::from(area.height);
let max_scroll = lines.len().saturating_sub(viewport);
if state.follow_selection {
if let Some(selected) = state.selected.as_deref() {
let first = lines
.iter()
.position(|line| line.row_id.as_deref() == Some(selected));
let last = lines
.iter()
.rposition(|line| line.row_id.as_deref() == Some(selected));
if let (Some(first), Some(last)) = (first, last) {
if first < state.scroll {
state.scroll = first;
} else if last >= state.scroll.saturating_add(viewport) {
state.scroll = last.saturating_add(1).saturating_sub(viewport);
}
}
} else {
state.scroll = max_scroll;
}
}
state.scroll = state.scroll.min(max_scroll);
let palette = palette::active();
for (offset, visual) in lines.iter().skip(state.scroll).take(viewport).enumerate() {
let selected = visual.row_id.as_ref() == state.selected.as_ref();
let surface = if selected {
palette.selection
} else {
palette.card
};
let foreground = match visual.tone {
Tone::Normal => surface.fg,
Tone::Muted => surface.muted,
Tone::Active => surface.active,
Tone::Warn => surface.awaiting,
Tone::Error => surface.error,
};
let base = Style::default().bg(surface.bg).fg(foreground);
let prefix = if visual.rail {
Span::styled("▌", base.fg(surface.awaiting))
} else {
Span::styled(" ", base)
};
let line = Line::from(vec![prefix, Span::styled(&visual.text, base)]);
f.render_widget(
Paragraph::new(line).style(base),
Rect::new(area.x, area.y + offset as u16, area.width, 1),
);
}
}
fn render_projection_status(&self, f: &mut Frame, area: Rect, state: &StructuredState) {
if area.height == 0 {
return;
}
let palette = palette::active();
let plan = plan_summary(state);
let usage = usage_summary(state);
let rows = Layout::default()
.direction(Direction::Vertical)
.constraints([Constraint::Length(1), Constraint::Length(1)])
.split(area);
f.render_widget(
Paragraph::new(plan).style(palette.card.muted_style()),
rows[0],
);
if area.height > 1 {
f.render_widget(
Paragraph::new(usage).style(palette.card.dim_style()),
rows[1],
);
}
}
fn render_composer(&self, f: &mut Frame, area: Rect, state: &StructuredState) {
if area.height == 0 {
return;
}
let palette = palette::active();
let active = state.turn_active();
let label = if active {
" working esc: cancel"
} else {
" › "
};
let label_width = label.chars().count().min(usize::from(area.width)) as u16;
let parts = Layout::default()
.direction(Direction::Horizontal)
.constraints([Constraint::Length(label_width), Constraint::Min(0)])
.split(area);
let label_style = if active {
Style::default().bg(palette.card.bg).fg(palette.card.active)
} else {
palette.card.text_style().add_modifier(Modifier::BOLD)
};
f.render_widget(Paragraph::new(label).style(label_style), parts[0]);
if !active {
let cursor = state
.composer
.render(parts[1], f.buffer_mut(), palette.card.text_style());
if self.focused && self.mode == Mode::Insert && parts[1].width > 0 {
f.set_cursor_position((parts[1].x + cursor, parts[1].y));
}
}
}
}
fn system_line(text: &str) -> VisualLine {
VisualLine {
text: text.into(),
row_id: None,
rail: false,
tone: Tone::Muted,
}
}
fn error_line(text: &str) -> VisualLine {
VisualLine {
text: text.into(),
row_id: None,
rail: true,
tone: Tone::Error,
}
}
fn build_lines(state: &StructuredState, width: usize) -> Vec<VisualLine> {
let mut out = Vec::new();
for row in &state.rows {
if row.kind == TranscriptRowKind::Plan {
continue;
}
let row_width = width.saturating_sub(1).max(1);
match row.kind {
TranscriptRowKind::User => push_wrapped(
&mut out,
row,
"You: ",
content_text(row),
row_width,
Tone::Normal,
false,
),
TranscriptRowKind::AgentText => push_wrapped(
&mut out,
row,
"",
content_text(row),
row_width,
Tone::Normal,
false,
),
TranscriptRowKind::AgentThought => push_wrapped(
&mut out,
row,
"thought: ",
content_text(row),
row_width,
Tone::Muted,
false,
),
TranscriptRowKind::ToolCall | TranscriptRowKind::ToolDiff => {
tool_lines(&mut out, state, row, row_width)
}
TranscriptRowKind::Approval => approval_lines(&mut out, state, row, row_width),
TranscriptRowKind::Notice => push_wrapped(
&mut out,
row,
"",
"[notice]".into(),
row_width,
Tone::Muted,
false,
),
TranscriptRowKind::Plan => {}
}
}
out
}
fn push_wrapped(
out: &mut Vec<VisualLine>,
row: &TranscriptRow,
prefix: &str,
text: String,
width: usize,
tone: Tone,
rail: bool,
) {
let joined = format!("{prefix}{text}");
for line in wrap_text(&joined, width) {
out.push(VisualLine {
text: line,
row_id: Some(row.id.clone()),
rail,
tone,
});
}
}
fn tool_lines(
out: &mut Vec<VisualLine>,
state: &StructuredState,
row: &TranscriptRow,
width: usize,
) {
let expanded = state.expanded.contains(&row.id);
let status = row
.data
.get("status")
.and_then(serde_json::Value::as_str)
.unwrap_or("completed");
let status_mark = match status {
"pending" => "·",
"in_progress" => "●",
"failed" => "×",
_ => "✓",
};
let (name, argument) = tool_title(row);
let duration = state
.tool_duration(&row.id)
.map(format_duration)
.unwrap_or_else(|| "—".into());
let affordance = if expanded { "▾" } else { "▸" };
let head = format!("{affordance} {status_mark} {name} {argument} {duration}");
out.push(VisualLine {
text: truncate_cells(&head, width),
row_id: Some(row.id.clone()),
rail: false,
tone: if status == "failed" {
Tone::Error
} else {
Tone::Active
},
});
let content = tool_content_lines(row);
if row.kind == TranscriptRowKind::ToolDiff && !expanded {
let shown = content.len().min(DIFF_PREVIEW_LINES);
for text in content.iter().take(shown) {
push_wrapped(out, row, " ", text.clone(), width, diff_tone(text), false);
}
if content.len() > shown {
push_wrapped(
out,
row,
" ",
"… Enter to expand".into(),
width,
Tone::Muted,
false,
);
}
} else if expanded {
for text in content {
push_wrapped(out, row, " ", text.clone(), width, diff_tone(&text), false);
}
}
}
fn approval_lines(
out: &mut Vec<VisualLine>,
state: &StructuredState,
row: &TranscriptRow,
width: usize,
) {
let pending = state.pending_nonce().is_some()
&& state
.rows
.iter()
.rev()
.find(|candidate| candidate.kind == TranscriptRowKind::Approval)
.is_some_and(|candidate| candidate.id == row.id);
let tool = row.data.get("toolCall").unwrap_or(&serde_json::Value::Null);
let title = tool
.get("title")
.and_then(serde_json::Value::as_str)
.unwrap_or("permission requested");
push_wrapped(
out,
row,
"permission: ",
title.into(),
width,
Tone::Warn,
true,
);
if let Some(path) = tool
.pointer("/locations/0/path")
.and_then(serde_json::Value::as_str)
{
push_wrapped(out, row, " ", path.into(), width, Tone::Muted, true);
}
let actions = if pending {
let offered = offered_decisions(&row.data);
let mut keys = Vec::new();
if offered.contains(&maestro_protocol::StructuredApprovalDecision::AllowOnce) {
keys.push("[a] allow once");
}
if offered.contains(&maestro_protocol::StructuredApprovalDecision::AllowForSession) {
keys.push("[A] allow session");
}
if offered.contains(&maestro_protocol::StructuredApprovalDecision::Deny) {
keys.push("[d] deny");
}
if keys.is_empty() {
"no decision offered".into()
} else {
keys.join(" ")
}
} else {
"answered".into()
};
push_wrapped(out, row, " ", actions, width, Tone::Warn, true);
}
fn content_text(row: &TranscriptRow) -> String {
let Some(content) = row.data.get("content") else {
return String::new();
};
match content.get("type").and_then(serde_json::Value::as_str) {
Some("text") => content
.get("text")
.and_then(serde_json::Value::as_str)
.unwrap_or_default()
.into(),
Some("image") => "[image]".into(),
Some("audio") => "[audio]".into(),
Some("resource_link") => format!(
"[resource: {}]",
content
.get("name")
.or_else(|| content.get("uri"))
.and_then(serde_json::Value::as_str)
.unwrap_or("linked")
),
Some("resource") => "[resource]".into(),
_ => String::new(),
}
}
fn tool_title(row: &TranscriptRow) -> (String, String) {
let fallback = row
.data
.get("kind")
.and_then(serde_json::Value::as_str)
.unwrap_or("tool");
let explicit = row.data.get("name").and_then(serde_json::Value::as_str);
let title = row
.data
.get("title")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
if let Some((name, rest)) = title
.split_once('(')
.and_then(|(name, rest)| rest.strip_suffix(')').map(|rest| (name, rest)))
{
return (explicit.unwrap_or(name).to_string(), rest.to_string());
}
(explicit.unwrap_or(fallback).to_string(), title.to_string())
}
fn tool_content_lines(row: &TranscriptRow) -> Vec<String> {
let mut out = Vec::new();
for block in row
.data
.get("content")
.and_then(serde_json::Value::as_array)
.into_iter()
.flatten()
{
match block.get("type").and_then(serde_json::Value::as_str) {
Some("content") => {
let nested = block.get("content").cloned().unwrap_or_default();
let fake = TranscriptRow {
id: String::new(),
kind: TranscriptRowKind::AgentText,
data: serde_json::json!({"content": nested}),
};
let text = content_text(&fake);
out.extend(text.lines().map(str::to_string));
}
Some("diff") => {
let path = block
.get("path")
.and_then(serde_json::Value::as_str)
.unwrap_or("file");
out.push(format!("{path}:"));
if let Some(old) = block.get("oldText").and_then(serde_json::Value::as_str) {
out.extend(old.lines().map(|line| format!("- {line}")));
}
if let Some(new) = block.get("newText").and_then(serde_json::Value::as_str) {
out.extend(new.lines().map(|line| format!("+ {line}")));
}
}
Some("terminal") => out.push("[embedded terminal]".into()),
_ => {}
}
}
out
}
fn diff_tone(text: &str) -> Tone {
if text.starts_with("+ ") {
Tone::Active
} else if text.starts_with("- ") {
Tone::Error
} else {
Tone::Muted
}
}
fn format_duration(duration: Duration) -> String {
if duration.as_secs() >= 60 {
format!("{}m{}s", duration.as_secs() / 60, duration.as_secs() % 60)
} else if duration.as_secs() > 0 {
format!("{}s", duration.as_secs())
} else {
format!("{}ms", duration.as_millis())
}
}
fn plan_summary(state: &StructuredState) -> String {
let entries = state
.reduced
.plan
.as_ref()
.and_then(|plan| plan.get("entries"))
.and_then(serde_json::Value::as_array);
let Some(entries) = entries else {
return " plan —".into();
};
let current = entries.iter().find(|entry| {
entry.get("status").and_then(serde_json::Value::as_str) == Some("in_progress")
});
let done = entries
.iter()
.filter(|entry| {
entry.get("status").and_then(serde_json::Value::as_str) == Some("completed")
})
.count();
let text = current
.and_then(|entry| entry.get("content"))
.and_then(serde_json::Value::as_str)
.unwrap_or("waiting");
format!(" plan {done}/{} {text}", entries.len())
}
fn usage_summary(state: &StructuredState) -> String {
let activity = if state.reduced.turn.thinking {
"thinking"
} else if state.reduced.turn.responding {
"responding"
} else {
"idle"
};
let usage = state
.reduced
.usage
.as_ref()
.or(state.reduced.context.as_ref());
let used = usage
.and_then(|value| value.get("used"))
.and_then(serde_json::Value::as_u64);
let size = usage
.and_then(|value| value.get("size"))
.and_then(serde_json::Value::as_u64);
let context = match (used, size) {
(Some(used), Some(size)) => format!(" context {used}/{size}"),
_ => String::new(),
};
let model = state
.reduced
.model
.as_deref()
.map(|model| format!(" {model}"))
.unwrap_or_default();
let mode = state
.reduced
.mode
.as_deref()
.map(|mode| format!(" · {mode}"))
.unwrap_or_default();
format!(" {activity}{context}{model}{mode}")
}
fn wrap_text(text: &str, width: usize) -> Vec<String> {
let width = width.max(1);
let mut out = Vec::new();
for source in text.split('\n') {
let mut line = String::new();
let mut cells = 0;
for ch in source.chars() {
let char_width = ch.width().unwrap_or(0);
if cells > 0 && cells + char_width > width {
out.push(std::mem::take(&mut line));
cells = 0;
}
line.push(ch);
cells += char_width;
}
out.push(line);
}
if out.is_empty() {
out.push(String::new());
}
out
}
fn truncate_cells(text: &str, width: usize) -> String {
let width = width.max(1);
let total: usize = text.chars().map(|ch| ch.width().unwrap_or(0)).sum();
if total <= width {
return text.to_string();
}
if width == 1 {
return "…".into();
}
let mut out = String::new();
let mut cells = 0;
for ch in text.chars() {
let char_width = ch.width().unwrap_or(0);
if cells + char_width > width - 1 {
break;
}
out.push(ch);
cells += char_width;
}
out.push('…');
out
}
#[cfg(test)]
mod tests {
use super::*;
use maestro_protocol::{ReducedState, TranscriptRow};
use ratatui::backend::TestBackend;
use ratatui::Terminal;
use serde_json::json;
fn row(id: &str, kind: TranscriptRowKind, data: serde_json::Value) -> TranscriptRow {
TranscriptRow {
id: id.into(),
kind,
data,
}
}
fn render_state(
state: StructuredState,
width: u16,
height: u16,
status: SessionStatus,
detail: Option<&str>,
) -> (String, ratatui::buffer::Buffer) {
let pane = StructuredPane {
handle: StructuredHandle::test_handle(state),
focused: true,
mode: Mode::Normal,
session_status: status,
activity_detail: detail.map(str::to_string),
};
let mut terminal = Terminal::new(TestBackend::new(width, height)).unwrap();
terminal
.draw(|frame| pane.render(frame, frame.area()))
.unwrap();
let buffer = terminal.backend().buffer().clone();
let text = (0..height)
.map(|y| {
(0..width)
.map(|x| buffer[(x, y)].symbol())
.collect::<String>()
})
.collect::<Vec<_>>()
.join("\n");
(text, buffer)
}
#[test]
fn prose_wraps_to_the_pane_width() {
let mut state = StructuredState::default();
state.rows.push(row(
"a",
TranscriptRowKind::AgentText,
json!({"content":{"type":"text","text":"abcdefghij"}}),
));
let lines = build_lines(&state, 5);
assert_eq!(
lines
.iter()
.map(|line| line.text.as_str())
.collect::<Vec<_>>(),
["abcd", "efgh", "ij"]
);
}
#[test]
fn tool_is_compact_until_selected_row_expands() {
let mut state = StructuredState::default();
state.rows.push(row("t", TranscriptRowKind::ToolCall, json!({"title":"Bash(cargo test)","status":"completed","content":[{"type":"content","content":{"type":"text","text":"output"}}]})));
let collapsed = build_lines(&state, 80);
assert_eq!(collapsed.len(), 1);
assert!(collapsed[0].text.contains("Bash cargo test"));
assert_eq!(build_lines(&state, 8).len(), 1);
state.selected = Some("t".into());
state.toggle_selected();
assert!(build_lines(&state, 80)
.iter()
.any(|line| line.text.contains("output")));
}
#[test]
fn collapsed_diff_previews_only_a_few_lines_with_affordance() {
let mut state = StructuredState::default();
state.rows.push(row("d", TranscriptRowKind::ToolDiff, json!({"title":"Edit(x)","content":[{"type":"diff","path":"x","oldText":"a\nb","newText":"c\nd"}]})));
let lines = build_lines(&state, 80);
assert!(lines.len() <= 5);
assert!(lines.last().unwrap().text.contains("Enter to expand"));
}
#[test]
fn plan_is_not_a_transcript_row_and_status_is_two_lines() {
let mut state = StructuredState::default();
state
.rows
.push(row("plan", TranscriptRowKind::Plan, json!({})));
state.reduced = ReducedState {
plan: Some(json!({"entries":[{"status":"in_progress","content":"ship"}]})),
usage: Some(json!({"used":12,"size":100})),
..ReducedState::default()
};
assert!(build_lines(&state, 80).is_empty());
assert!(plan_summary(&state).contains("ship"));
assert!(usage_summary(&state).contains("12/100"));
}
#[test]
fn approval_is_inline_railed_and_lists_only_offered_keys() {
let mut state = StructuredState::default();
state.rows.push(row("p", TranscriptRowKind::Approval, json!({"toolCall":{"title":"Bash(rm x)"},"options":[{"kind":"allow_once"},{"kind":"reject_once"}]})));
state.reduced.pending_approval = Some(json!({"nonce":"n"}));
let lines = build_lines(&state, 80);
assert!(lines.iter().all(|line| line.rail));
assert!(lines
.iter()
.any(|line| line.text.contains("[a] allow once")));
assert!(!lines.iter().any(|line| line.text.contains("[A]")));
}
#[test]
fn one_column_panes_do_not_panic() {
let mut state = StructuredState::default();
state.rows.push(row(
"a",
TranscriptRowKind::AgentText,
json!({"content":{"type":"text","text":"hi"}}),
));
assert!(!build_lines(&state, 1).is_empty());
}
#[test]
fn native_pane_renders_projection_footer_and_composer_together() {
let mut state = StructuredState::default();
state.rows.push(row(
"a",
TranscriptRowKind::AgentText,
json!({"content":{"type":"text","text":"projected transcript"}}),
));
state.reduced.plan =
Some(json!({"entries":[{"status":"in_progress","content":"render native pane"}]}));
state.reduced.usage = Some(json!({"used":42,"size":1000}));
let (text, _) = render_state(state, 50, 8, SessionStatus::Running, None);
assert!(text.contains("projected transcript"), "{text}");
assert!(text.contains("render native pane"), "{text}");
assert!(text.contains("context 42/1000"), "{text}");
assert!(text.contains('›'), "{text}");
}
#[test]
fn selected_row_scrolls_into_view_and_uses_the_selection_surface() {
let mut state = StructuredState::default();
for n in 0..12 {
state.rows.push(row(
&format!("r{n}"),
TranscriptRowKind::AgentText,
json!({"content":{"type":"text","text":format!("line {n}")}}),
));
}
state.selected = Some("r11".into());
let (text, buffer) = render_state(state, 30, 7, SessionStatus::Running, None);
assert!(text.contains("line 11"), "{text}");
let selected_y = (0..4)
.find(|y| {
(0..30)
.map(|x| buffer[(x, *y)].symbol())
.collect::<String>()
.contains("line 11")
})
.unwrap();
assert_eq!(buffer[(2, selected_y)].bg, palette::active().selection.bg);
}
#[test]
fn dead_session_reason_and_connection_error_are_rendered_inline() {
let mut state = StructuredState::default();
state.error = Some("request failed".into());
let (text, _) = render_state(state, 50, 7, SessionStatus::Dead, Some("adapter exited"));
assert!(text.contains("adapter exited"), "{text}");
assert!(text.contains("request failed"), "{text}");
}
}
crates/maestro-tui/src/structured_task.rs
+600
View File
@@ -0,0 +1,600 @@
//! Live projection state for one structured session.
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use futures_util::StreamExt;
use maestro_protocol::{
ReducedState, StructuredApprovalDecision, StructuredEvent, TranscriptDelta, TranscriptRow,
TranscriptRowKind,
};
use tokio::sync::watch;
use tokio_tungstenite::tungstenite::{Error as WsError, Message};
use crate::input::InputKey;
use crate::session_task::SessionState;
use crate::text_input::TextInput;
#[derive(Debug, Clone)]
struct ToolTiming {
started: Instant,
elapsed: Option<Duration>,
}
/// Renderer and input state built directly from the daemon's two projections.
#[derive(Debug, Clone)]
pub struct StructuredState {
pub rows: Vec<TranscriptRow>,
pub reduced: ReducedState,
pub selected: Option<String>,
pub expanded: HashSet<String>,
pub scroll: usize,
pub follow_selection: bool,
pub composer: TextInput,
pub error: Option<String>,
submitting: bool,
approval_submitting: bool,
tool_timings: HashMap<String, ToolTiming>,
}
impl Default for StructuredState {
fn default() -> Self {
Self {
rows: Vec::new(),
reduced: ReducedState::default(),
selected: None,
expanded: HashSet::new(),
scroll: 0,
follow_selection: true,
composer: TextInput::default(),
error: None,
submitting: false,
approval_submitting: false,
tool_timings: HashMap::new(),
}
}
}
impl StructuredState {
pub fn apply(&mut self, event: StructuredEvent) {
match event {
StructuredEvent::TranscriptSnapshot { snapshot } => {
let selected = self.selected.clone();
self.rows = snapshot.rows;
self.selected = selected
.filter(|id| self.rows.iter().any(|row| &row.id == id))
.or_else(|| {
self.rows
.iter()
.rev()
.find(|row| row.kind != TranscriptRowKind::Plan)
.map(|row| row.id.clone())
});
self.tool_timings.clear();
let rows = self.rows.clone();
for row in &rows {
self.observe_tool(row);
}
self.expanded
.retain(|id| self.rows.iter().any(|row| &row.id == id));
}
StructuredEvent::ReducedState { state } => self.apply_reduced(state),
StructuredEvent::ProjectionUpdate { transcript, state } => {
if let Some(delta) = transcript {
self.apply_delta(delta);
}
self.apply_reduced(state);
}
}
}
fn apply_delta(&mut self, delta: TranscriptDelta) {
match delta {
TranscriptDelta::Append { row } => {
let followed_tail = self.follow_selection
&& self.selected.as_ref().is_some_and(|id| {
self.rows
.iter()
.rev()
.find(|old| old.kind != TranscriptRowKind::Plan)
.is_some_and(|old| &old.id == id)
});
if let Some(index) = self.rows.iter().position(|old| old.id == row.id) {
self.rows[index] = row.clone();
} else {
self.rows.push(row.clone());
}
self.observe_tool(&row);
if row.kind != TranscriptRowKind::Plan && (self.selected.is_none() || followed_tail)
{
self.selected = Some(row.id);
}
}
TranscriptDelta::Patch { row } => {
if let Some(index) = self.rows.iter().position(|old| old.id == row.id) {
self.rows[index] = row.clone();
} else {
self.rows.push(row.clone());
}
self.observe_tool(&row);
}
TranscriptDelta::Remove { id } => {
let removed = self.rows.iter().position(|row| row.id == id);
self.rows.retain(|row| row.id != id);
self.expanded.remove(&id);
self.tool_timings.remove(&id);
if self.selected.as_deref() == Some(&id) {
self.selected = removed.and_then(|index| {
let split = index.min(self.rows.len());
self.rows[split..]
.iter()
.find(|row| row.kind != TranscriptRowKind::Plan)
.or_else(|| {
self.rows[..split]
.iter()
.rev()
.find(|row| row.kind != TranscriptRowKind::Plan)
})
.map(|row| row.id.clone())
});
}
}
}
}
fn apply_reduced(&mut self, state: ReducedState) {
let previous_nonce = self.pending_nonce().map(str::to_owned);
if state.turn.responding || state.turn.thinking {
self.submitting = false;
}
self.reduced = state;
let next_nonce = self.pending_nonce().map(str::to_owned);
if next_nonce != previous_nonce {
self.approval_submitting = false;
if next_nonce.is_some() {
self.selected = self
.rows
.iter()
.rev()
.find(|row| row.kind == TranscriptRowKind::Approval)
.map(|row| row.id.clone());
self.follow_selection = true;
}
}
}
fn observe_tool(&mut self, row: &TranscriptRow) {
if !matches!(
row.kind,
TranscriptRowKind::ToolCall | TranscriptRowKind::ToolDiff
) {
return;
}
let active = matches!(
row.data.get("status").and_then(serde_json::Value::as_str),
Some("pending" | "in_progress")
);
let now = Instant::now();
let timing = self
.tool_timings
.entry(row.id.clone())
.or_insert(ToolTiming {
started: now,
elapsed: None,
});
if !active && timing.elapsed.is_none() {
timing.elapsed = Some(now.saturating_duration_since(timing.started));
}
}
pub fn tool_duration(&self, id: &str) -> Option<Duration> {
self.tool_timings.get(id).map(|timing| {
timing
.elapsed
.unwrap_or_else(|| Instant::now().saturating_duration_since(timing.started))
})
}
pub fn select(&mut self, delta: isize) {
let selectable: Vec<&TranscriptRow> = self
.rows
.iter()
.filter(|row| row.kind != TranscriptRowKind::Plan)
.collect();
if selectable.is_empty() {
self.selected = None;
return;
}
let current = self
.selected
.as_ref()
.and_then(|id| selectable.iter().position(|row| &row.id == id))
.unwrap_or(selectable.len() - 1);
let next = if delta >= 0 {
current
.saturating_add(delta as usize)
.min(selectable.len() - 1)
} else {
current.saturating_sub(delta.unsigned_abs())
};
self.selected = Some(selectable[next].id.clone());
self.follow_selection = true;
}
pub fn toggle_selected(&mut self) {
let Some(id) = self.selected.clone() else {
return;
};
let expandable = self.rows.iter().any(|row| {
row.id == id
&& matches!(
row.kind,
TranscriptRowKind::ToolCall | TranscriptRowKind::ToolDiff
)
});
if !expandable {
return;
}
if !self.expanded.insert(id.clone()) {
self.expanded.remove(&id);
}
self.follow_selection = true;
}
pub fn scroll_by(&mut self, lines: isize) {
self.scroll = if lines >= 0 {
self.scroll.saturating_sub(lines as usize)
} else {
self.scroll.saturating_add(lines.unsigned_abs())
};
self.follow_selection = false;
}
pub fn edit_composer(&mut self, key: &InputKey) {
if !self.turn_active() {
self.composer.handle_key(key);
self.error = None;
}
}
pub fn begin_prompt(&mut self) -> Option<String> {
if self.turn_active() || self.composer.value().trim().is_empty() {
return None;
}
let prompt = self.composer.value().to_string();
self.composer = TextInput::default();
self.submitting = true;
self.error = None;
Some(prompt)
}
pub fn prompt_failed(&mut self, prompt: String, error: String) {
if self.composer.value().is_empty() {
self.composer = TextInput::with_value(prompt);
}
self.submitting = false;
self.error = Some(error);
}
pub fn turn_active(&self) -> bool {
self.submitting || self.reduced.turn.responding || self.reduced.turn.thinking
}
pub fn pending_nonce(&self) -> Option<&str> {
self.reduced
.pending_approval
.as_ref()?
.get("nonce")?
.as_str()
}
pub fn approval_decision(&mut self, key: char) -> Option<(String, StructuredApprovalDecision)> {
if self.approval_submitting {
return None;
}
let selected = self.selected.as_deref()?;
let pending_row = self
.rows
.iter()
.rev()
.find(|row| row.kind == TranscriptRowKind::Approval)?;
if pending_row.id != selected {
return None;
}
let decision = match key {
'a' => StructuredApprovalDecision::AllowOnce,
'A' => StructuredApprovalDecision::AllowForSession,
'd' => StructuredApprovalDecision::Deny,
_ => return None,
};
let offered = offered_decisions(&pending_row.data);
if !offered.contains(&decision) {
return None;
}
let nonce = self.pending_nonce()?.to_string();
self.approval_submitting = true;
self.error = None;
Some((nonce, decision))
}
pub fn approval_failed(&mut self, error: String) {
self.approval_submitting = false;
self.error = Some(error);
}
}
pub fn offered_decisions(data: &serde_json::Value) -> Vec<StructuredApprovalDecision> {
let mut out = Vec::new();
for kind in data
.get("options")
.and_then(serde_json::Value::as_array)
.into_iter()
.flatten()
.filter_map(|option| option.get("kind").and_then(serde_json::Value::as_str))
{
match kind {
"allow_once" if !out.contains(&StructuredApprovalDecision::AllowOnce) => {
out.push(StructuredApprovalDecision::AllowOnce)
}
"allow_always" if !out.contains(&StructuredApprovalDecision::AllowForSession) => {
out.push(StructuredApprovalDecision::AllowForSession)
}
"reject_once" | "reject_always" if !out.contains(&StructuredApprovalDecision::Deny) => {
out.push(StructuredApprovalDecision::Deny)
}
_ => {}
}
}
out
}
/// Cheap shared handle for a structured projection connection.
#[derive(Clone)]
pub struct StructuredHandle {
pub state: Arc<Mutex<StructuredState>>,
connection: Arc<AtomicU8>,
revision: Arc<AtomicU64>,
stop: watch::Sender<bool>,
}
impl std::fmt::Debug for StructuredHandle {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("StructuredHandle")
.field("connection", &self.connection())
.field("revision", &self.revision())
.finish_non_exhaustive()
}
}
impl StructuredHandle {
pub fn connection(&self) -> SessionState {
match self.connection.load(Ordering::Acquire) {
0 => SessionState::Connecting,
1 => SessionState::Live,
2 => SessionState::Reconnecting,
_ => SessionState::Ended,
}
}
pub fn revision(&self) -> u64 {
self.revision.load(Ordering::Acquire)
}
pub fn end(&self) {
self.connection
.store(SessionState::Ended as u8, Ordering::Release);
let _ = self.stop.send(true);
}
#[cfg(test)]
pub(crate) fn test_handle(state: StructuredState) -> Self {
let (stop, _) = watch::channel(false);
Self {
state: Arc::new(Mutex::new(state)),
connection: Arc::new(AtomicU8::new(SessionState::Live as u8)),
revision: Arc::new(AtomicU64::new(0)),
stop,
}
}
}
#[derive(Debug)]
pub struct StructuredTask;
impl StructuredTask {
pub fn spawn(ws_url: String) -> StructuredHandle {
let state = Arc::new(Mutex::new(StructuredState::default()));
let connection = Arc::new(AtomicU8::new(SessionState::Connecting as u8));
let revision = Arc::new(AtomicU64::new(0));
let (stop, mut stopped) = watch::channel(false);
let handle = StructuredHandle {
state: state.clone(),
connection: connection.clone(),
revision: revision.clone(),
stop,
};
tokio::spawn(async move {
let mut backoff = Duration::from_secs(1);
loop {
if *stopped.borrow() {
return;
}
connection.store(SessionState::Connecting as u8, Ordering::Release);
let connected = tokio::select! {
result = tokio_tungstenite::connect_async(&ws_url) => result,
_ = stopped.changed() => return,
};
let (mut ws, _) = match connected {
Ok(value) => value,
Err(error) => {
if terminal_handshake_error(&error) {
state.lock().expect("structured state mutex").error =
Some(format!("cannot open structured session: {error}"));
connection.store(SessionState::Ended as u8, Ordering::Release);
revision.fetch_add(1, Ordering::Release);
return;
}
connection.store(SessionState::Reconnecting as u8, Ordering::Release);
revision.fetch_add(1, Ordering::Release);
tokio::select! {
_ = tokio::time::sleep(super::session_task::jitter(backoff)) => {},
_ = stopped.changed() => return,
}
backoff = backoff.saturating_mul(2).min(Duration::from_secs(30));
continue;
}
};
backoff = Duration::from_secs(1);
connection.store(SessionState::Live as u8, Ordering::Release);
revision.fetch_add(1, Ordering::Release);
loop {
let frame = tokio::select! {
frame = ws.next() => frame,
_ = stopped.changed() => return,
};
match frame {
Some(Ok(Message::Text(text))) => {
if let Ok(event) = serde_json::from_str::<StructuredEvent>(&text) {
state.lock().expect("structured state mutex").apply(event);
revision.fetch_add(1, Ordering::Release);
}
}
Some(Ok(Message::Close(_))) | None | Some(Err(_)) => break,
_ => {}
}
}
connection.store(SessionState::Reconnecting as u8, Ordering::Release);
revision.fetch_add(1, Ordering::Release);
tokio::select! {
_ = tokio::time::sleep(super::session_task::jitter(backoff)) => {},
_ = stopped.changed() => return,
}
backoff = backoff.saturating_mul(2).min(Duration::from_secs(30));
}
});
handle
}
}
fn terminal_handshake_error(error: &WsError) -> bool {
matches!(error, WsError::Http(response) if response.status().is_client_error())
}
#[cfg(test)]
mod tests {
use super::*;
use maestro_protocol::{TranscriptSnapshot, TurnState};
use serde_json::json;
fn row(id: &str, kind: TranscriptRowKind, data: serde_json::Value) -> TranscriptRow {
TranscriptRow {
id: id.into(),
kind,
data,
}
}
#[test]
fn snapshots_replace_and_deltas_address_rows_by_id() {
let mut state = StructuredState::default();
state.apply(StructuredEvent::TranscriptSnapshot {
snapshot: TranscriptSnapshot {
cursor: 2,
rows: vec![row("a", TranscriptRowKind::AgentText, json!({}))],
},
});
state.apply(StructuredEvent::ProjectionUpdate {
transcript: Some(TranscriptDelta::Patch {
row: row("a", TranscriptRowKind::AgentText, json!({"n": 2})),
}),
state: ReducedState::default(),
});
state.apply(StructuredEvent::ProjectionUpdate {
transcript: Some(TranscriptDelta::Append {
row: row("b", TranscriptRowKind::Notice, json!({})),
}),
state: ReducedState::default(),
});
assert_eq!(state.rows.len(), 2);
assert_eq!(state.rows[0].data["n"], 2);
state.apply(StructuredEvent::TranscriptSnapshot {
snapshot: TranscriptSnapshot {
cursor: 5,
rows: vec![row("b", TranscriptRowKind::Notice, json!({"fresh": true}))],
},
});
assert_eq!(state.rows.len(), 1);
assert_eq!(state.rows[0].data["fresh"], true);
}
#[test]
fn selection_survives_patch_and_moves_after_remove() {
let mut state = StructuredState {
rows: vec![
row("a", TranscriptRowKind::Notice, json!({})),
row("b", TranscriptRowKind::ToolCall, json!({})),
row("plan", TranscriptRowKind::Plan, json!({})),
],
selected: Some("a".into()),
..StructuredState::default()
};
state.select(1);
assert_eq!(state.selected.as_deref(), Some("b"));
state.toggle_selected();
assert!(state.expanded.contains("b"));
state.apply_delta(TranscriptDelta::Remove { id: "b".into() });
assert_eq!(state.selected.as_deref(), Some("a"));
state.apply_delta(TranscriptDelta::Remove { id: "a".into() });
assert_eq!(
state.selected, None,
"collapsed plan rows are not selectable"
);
}
#[test]
fn approval_keys_only_offer_server_backed_decisions() {
let mut state = StructuredState::default();
state.rows.push(row(
"approval:1",
TranscriptRowKind::Approval,
json!({"options":[
{"kind":"allow_once"}, {"kind":"reject_always"}
]}),
));
state.selected = Some("approval:1".into());
state.reduced.pending_approval = Some(json!({"nonce":"secret"}));
assert_eq!(
state.approval_decision('A'),
None,
"allow-for-session was not offered"
);
assert_eq!(
state.approval_decision('a'),
Some(("secret".into(), StructuredApprovalDecision::AllowOnce))
);
}
#[test]
fn composer_is_gated_by_the_projected_turn_lifecycle() {
let mut state = StructuredState::default();
state.edit_composer(&InputKey::char('h'));
state.edit_composer(&InputKey::char('i'));
assert_eq!(state.begin_prompt().as_deref(), Some("hi"));
assert!(state.turn_active());
state.prompt_failed("hi".into(), "nope".into());
assert_eq!(state.composer.value(), "hi");
state.reduced.turn = TurnState {
responding: true,
..TurnState::default()
};
state.edit_composer(&InputKey::char('x'));
assert_eq!(state.composer.value(), "hi");
}
}
crates/maestro-tui/tests/structured_http.rs
+79
View File
@@ -0,0 +1,79 @@
//! The TUI's structured actions use the shared typed daemon client; assert the
//! exact route and bare approval decision body at the HTTP boundary.
use std::sync::{Arc, Mutex};
use axum::extract::{Path, State};
use axum::routing::post;
use axum::{Json, Router};
use maestro_protocol::{DaemonClient, StructuredApprovalDecision, StructuredPromptRequest};
use uuid::Uuid;
#[derive(Debug, Default)]
struct Seen {
prompt: Option<String>,
cancelled: bool,
approval: Option<(String, StructuredApprovalDecision)>,
}
type Shared = Arc<Mutex<Seen>>;
async fn prompt(
State(seen): State<Shared>,
Path(_id): Path<Uuid>,
Json(body): Json<StructuredPromptRequest>,
) {
seen.lock().unwrap().prompt = Some(body.prompt);
}
async fn cancel(State(seen): State<Shared>, Path(_id): Path<Uuid>) {
seen.lock().unwrap().cancelled = true;
}
async fn approval(
State(seen): State<Shared>,
Path((_id, nonce)): Path<(Uuid, String)>,
Json(decision): Json<StructuredApprovalDecision>,
) {
seen.lock().unwrap().approval = Some((nonce, decision));
}
#[tokio::test(flavor = "multi_thread")]
async fn prompt_cancel_and_approval_use_structured_api_contract() {
let seen = Shared::default();
let app = Router::new()
.route("/api/sessions/:id/structured/prompt", post(prompt))
.route("/api/sessions/:id/structured/cancel", post(cancel))
.route(
"/api/sessions/:id/structured/approvals/:nonce",
post(approval),
)
.with_state(seen.clone());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = DaemonClient::new(format!("http://{addr}"));
let id = Uuid::from_u128(141);
client.send_structured_prompt(id, "ship it").await.unwrap();
client.cancel_structured_turn(id).await.unwrap();
client
.resolve_structured_approval(
id,
"one-use-nonce",
StructuredApprovalDecision::AllowForSession,
)
.await
.unwrap();
let seen = seen.lock().unwrap();
assert_eq!(seen.prompt.as_deref(), Some("ship it"));
assert!(seen.cancelled);
assert_eq!(
seen.approval,
Some((
"one-use-nonce".into(),
StructuredApprovalDecision::AllowForSession
))
);
}
crates/maestro-tui/tests/structured_task_ws.rs
+107
View File
@@ -0,0 +1,107 @@
//! Projection socket coverage: snapshots replace state on reconnect and live
//! deltas patch rows without client-side event folding.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use axum::extract::ws::{Message, WebSocketUpgrade};
use axum::extract::State;
use axum::response::Response;
use axum::routing::any;
use axum::Router;
use maestro_protocol::{
ReducedState, StructuredEvent, TranscriptDelta, TranscriptRow, TranscriptRowKind,
TranscriptSnapshot,
};
use maestro_tui::session_task::SessionState;
use maestro_tui::structured_task::StructuredTask;
use serde_json::json;
async fn structured_handler(
State(connects): State<Arc<AtomicUsize>>,
ws: WebSocketUpgrade,
) -> Response {
ws.on_upgrade(move |mut socket| async move {
let attempt = connects.fetch_add(1, Ordering::SeqCst);
let snapshot = if attempt == 0 {
TranscriptSnapshot {
cursor: 1,
rows: vec![row("old", json!({"content":{"type":"text","text":"old"}}))],
}
} else {
TranscriptSnapshot {
cursor: 3,
rows: vec![row("live", json!({"content":{"type":"text","text":"one"}}))],
}
};
send(
&mut socket,
StructuredEvent::TranscriptSnapshot { snapshot },
)
.await;
if attempt == 0 {
let _ = socket.close().await;
return;
}
send(
&mut socket,
StructuredEvent::ProjectionUpdate {
transcript: Some(TranscriptDelta::Patch {
row: row("live", json!({"content":{"type":"text","text":"two"}})),
}),
state: ReducedState {
model: Some("test-model".into()),
..ReducedState::default()
},
},
)
.await;
tokio::time::sleep(Duration::from_secs(3)).await;
})
}
fn row(id: &str, data: serde_json::Value) -> TranscriptRow {
TranscriptRow {
id: id.into(),
kind: TranscriptRowKind::AgentText,
data,
}
}
async fn send(socket: &mut axum::extract::ws::WebSocket, event: StructuredEvent) {
socket
.send(Message::Text(serde_json::to_string(&event).unwrap()))
.await
.unwrap();
}
#[tokio::test(flavor = "multi_thread")]
async fn reconnect_snapshot_replaces_then_delta_patches_projection() {
let connects = Arc::new(AtomicUsize::new(0));
let app = Router::new()
.route("/api/structured/x", any(structured_handler))
.with_state(connects.clone());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let handle = StructuredTask::spawn(format!("ws://{addr}/api/structured/x"));
for _ in 0..250 {
tokio::time::sleep(Duration::from_millis(10)).await;
let state = handle.state.lock().unwrap().clone();
if state.reduced.model.as_deref() == Some("test-model") {
assert_eq!(state.rows.len(), 1);
assert_eq!(state.rows[0].id, "live");
assert_eq!(state.rows[0].data["content"]["text"], "two");
assert!(connects.load(Ordering::SeqCst) >= 2);
handle.end();
assert_eq!(handle.connection(), SessionState::Ended);
return;
}
}
panic!(
"structured projection never reached reconnect delta; connection={:?}",
handle.connection()
);
}