Compare commits

..
Author SHA1 Message Date
Miguel Palhas 24ed1ac002 docs(maestro): fix two more factual errors in structured docs
INSTALL.md conflated event-log replay with ACP session/load resume.
STRUCTURED-VIEW.md overstated death-on-append-failure: only notification
appends kill the session, approval-row append failures just cancel the
pending approval.
2026-09-05 03:49:59 +01:00
Miguel PalhasandClaude Sonnet 5 64741a027e docs(maestro): fix five more factual errors in structured-session docs
Corrects recovery-as-two-operations, tmux-only survival on restart,
replay's durable-vs-live-state gap, snapshot-vs-replay purpose, and the
three-file persistence count.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-05 03:45:26 +01:00
Miguel Palhas cdd4213ff5 docs(maestro): fix three factual errors in STRUCTURED-VIEW.md
- replay: since=0 returns the first page (still limit-bounded), not the
  whole log; clients page forward from the last returned cursor
- PR_SET_PDEATHSIG is Linux-only; other platforms depend on the
  terminate() ladder running to completion
- approval nonce is minted before cancelling the prior pending approval,
  not after
2026-09-05 03:38:32 +01:00
Miguel Palhas f8973b3802 docs(maestro): write STRUCTURED-VIEW.md
Contracts the eight transcript row kinds, daemon-vs-client decisions, the
transcript/reduced-state projections and their cursor/replay semantics, the
ACP handshake and declined capabilities, the adapter install command, and
the approval nonce rules (generation, live-only placement, secrecy from the
agent and durable history, single-use/duplicate resolution) and bounded
prompt/cancel liveness.

Amends RESTART.md to state plainly that a structured session's agent is a
daemon child, not tmux-backed: a restart ends the active turn, the
transcript survives in the event log, and pending approvals/nonces do not.

Refs #137
2026-09-05 03:32:42 +01:00
Miguel Palhas 1e2c0f0a38 Merge structured-view/139-permission-requests into blitz/structured-view 2026-09-05 03:14:19 +01:00
Miguel Palhas 6fa7dc63e4 fix: serialize structured prompt dispatch 2026-09-05 03:11:16 +01:00
Miguel Palhas 5c0a4c4640 fix: bind approvals to prompt turns 2026-09-05 02:57:29 +01:00
Miguel Palhas 8773671afd fix(daemon): unify approval replay log 2026-09-05 02:44:59 +01:00
Miguel Palhas d9815d86c6 feat(daemon): resolve ACP approvals 2026-09-05 02:32:01 +01:00
Miguel Palhas 13009d0f0e Merge structured-view/138-structured-api into blitz/structured-view 2026-09-05 02:16:00 +01:00
15 changed files with 1572 additions and 100 deletions
+6
View File
@@ -92,6 +92,12 @@ clients — how it opens, the sigils, the key model, the match-and-rank algorith
the two matchers must agree on, and the fixture table they are tested against.
Read it before changing either bar or either matcher.
`docs/STRUCTURED-VIEW.md` is the contract for a structured session (an agent
run as a daemon child over ACP instead of in a tmux pane) — the eight
transcript row kinds, the two projections and their cursor/replay semantics,
the handshake and declined capabilities, and the approval nonce rules. Read
it before changing anything that renders a structured session.
`nix run .#dev` (or `just dev`) brings up the whole local stack under
process-compose: a debug daemon on 127.0.0.1:7071, then Vite on :5173 once the
daemon answers `/api/info`. The SPA proxies `/api/*` back to the daemon, so only
+100 -11
View File
@@ -113,6 +113,15 @@ impl SpawnConfig {
}
}
/// A blocked ACP permission request delivered to the daemon.
#[derive(Debug)]
pub struct PermissionRequest {
/// The prompt turn that was active when ACP delivered this callback.
pub turn: u64,
pub request: RequestPermissionRequest,
pub response: oneshot::Sender<RequestPermissionOutcome>,
}
#[derive(Debug, thiserror::Error)]
pub enum AcpError {
/// The adapter is not on `PATH`. The message names the install command
@@ -156,6 +165,7 @@ enum Command {
Cancel,
Prompt {
text: String,
turn: u64,
done: oneshot::Sender<Result<(), String>>,
},
Close,
@@ -166,6 +176,31 @@ enum ChildCommand {
Reap(oneshot::Sender<bool>),
}
#[derive(Debug, Default)]
struct PermissionTurnState {
active: Option<u64>,
}
impl PermissionTurnState {
fn begin(&mut self, turn: u64) {
self.active = Some(turn);
}
fn cancel_active(&mut self) {
self.active = None;
}
fn finish(&mut self, turn: u64) {
if self.active == Some(turn) {
self.active = None;
}
}
fn callback_turn(&self) -> Option<u64> {
self.active
}
}
/// Why the handshake did not produce a session id.
enum ReadyError {
/// The adapter answered with an error.
@@ -192,6 +227,7 @@ pub struct AcpSession {
commands: mpsc::UnboundedSender<Command>,
child_commands: mpsc::UnboundedSender<ChildCommand>,
updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
exit: watch::Receiver<Option<ExitStatus>>,
}
@@ -235,6 +271,13 @@ impl AcpSession {
self.updates.lock().expect("updates mutex").take()
}
pub fn take_permission_requests(&self) -> Option<mpsc::UnboundedReceiver<PermissionRequest>> {
self.permission_requests
.lock()
.expect("permission requests")
.take()
}
/// Send `session/cancel`. Fire-and-forget: a closed connection means the
/// child is already gone, which is the outcome cancel was after.
pub fn cancel(&self) {
@@ -242,11 +285,16 @@ impl AcpSession {
}
/// Send a user prompt and receive completion once the agent ends the turn.
pub fn prompt(&self, text: String) -> Result<oneshot::Receiver<Result<(), String>>, String> {
pub fn prompt(
&self,
text: String,
turn: u64,
) -> Result<oneshot::Receiver<Result<(), String>>, String> {
let (done_tx, done_rx) = oneshot::channel();
self.commands
.send(Command::Prompt {
text,
turn,
done: done_tx,
})
.map_err(|_| "ACP connection is closed".to_string())?;
@@ -335,6 +383,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
let (updates_tx, updates_rx) = mpsc::unbounded_channel();
let (permission_tx, permission_rx) = mpsc::unbounded_channel();
let (commands_tx, commands_rx) = mpsc::unbounded_channel();
let (ready_tx, ready_rx) = oneshot::channel();
let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
@@ -343,6 +392,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
config.cwd.clone(),
config.load_session.clone(),
updates_tx,
permission_tx,
commands_rx,
ready_tx,
));
@@ -356,6 +406,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
commands: commands_tx,
child_commands: child_commands_tx,
updates: Mutex::new(Some(updates_rx)),
permission_requests: Mutex::new(Some(permission_rx)),
exit: exit_rx,
});
}
@@ -467,6 +518,7 @@ async fn run_connection<T>(
cwd: PathBuf,
load_session: Option<String>,
updates_tx: mpsc::UnboundedSender<SessionNotification>,
permission_requests: mpsc::UnboundedSender<PermissionRequest>,
mut commands_rx: mpsc::UnboundedReceiver<Command>,
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
) where
@@ -478,6 +530,10 @@ async fn run_connection<T>(
// history the event log already holds (#135).
let replaying = Arc::new(AtomicBool::new(false));
let replaying_seen = replaying.clone();
// ACP callbacks have no turn id. The daemon never queues a later prompt
// until a cancelled prompt settles, so this active turn is unambiguous.
let permission_turn = Arc::new(Mutex::new(PermissionTurnState::default()));
let permission_turn_seen = permission_turn.clone();
let result = Client
.builder()
.name("maestro")
@@ -495,15 +551,30 @@ async fn run_connection<T>(
)
.on_receive_request(
async move |request: RequestPermissionRequest, responder, _cx| {
// Stub until #139: decline rather than grant something the
// operator never saw.
tracing::warn!(
session = %request.session_id,
"session/request_permission declined: not implemented yet (#139)"
);
responder.respond(RequestPermissionResponse::new(
RequestPermissionOutcome::Cancelled,
))
let turn = permission_turn_seen
.lock()
.expect("permission turn state")
.callback_turn();
let Some(turn) = turn else {
return responder.respond(RequestPermissionResponse::new(
RequestPermissionOutcome::Cancelled,
));
};
let (response, received) = oneshot::channel();
if permission_requests
.send(PermissionRequest {
turn,
request,
response,
})
.is_err()
{
return responder.respond(RequestPermissionResponse::new(
RequestPermissionOutcome::Cancelled,
));
}
let outcome = received.await.unwrap_or(RequestPermissionOutcome::Cancelled);
responder.respond(RequestPermissionResponse::new(outcome))
},
agent_client_protocol::on_receive_request!(),
)
@@ -575,26 +646,44 @@ async fn run_connection<T>(
tokio::select! {
cmd = commands_rx.recv() => match cmd {
Some(Command::Cancel) => {
permission_turn
.lock()
.expect("permission turn state")
.cancel_active();
let _ = cx.send_notification(CancelNotification::new(
SessionId::new(session_id.as_str()),
));
}
Some(Command::Prompt { text, done }) => {
Some(Command::Prompt { text, turn, done }) => {
permission_turn
.lock()
.expect("permission turn state")
.begin(turn);
let request = PromptRequest::new(
SessionId::new(session_id.as_str()),
vec![ContentBlock::from(text)],
);
let done = Arc::new(Mutex::new(Some(done)));
let callback_done = done.clone();
let callback_turn = permission_turn.clone();
if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
let callback_done = callback_done.clone();
let callback_turn = callback_turn.clone();
async move {
callback_turn
.lock()
.expect("permission turn state")
.finish(turn);
if let Some(done) = callback_done.lock().expect("prompt sender").take() {
let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
}
Ok(())
}
}) {
permission_turn
.lock()
.expect("permission turn state")
.finish(turn);
if let Some(done) = done.lock().expect("prompt sender").take() {
let _ = done.send(Err(error.to_string()));
}
@@ -24,6 +24,7 @@ Environment knobs, all optional:
so the SIGTERM path is exercised)
FAKE_ACP_IGNORE_EOF=1 keep running after stdin closes, until signalled
FAKE_ACP_EMIT_ON_CANCEL=1 emit a final notification before handling cancel
FAKE_ACP_CANCEL_SETTLES=1 complete the outstanding prompt on cancel
FAKE_ACP_NO_LOAD=1 advertise `loadSession: false`
FAKE_ACP_LOAD_REPLAY=n on `session/load`, replay n `agent_message_chunk`
notifications ("replayed <i>") before answering,
@@ -33,6 +34,8 @@ Environment knobs, all optional:
FAKE_ACP_PROMPT_FAIL=1 answer every `session/prompt` with a JSON-RPC error
FAKE_ACP_PROMPT_FAIL_ONCE=1 answer only the first `session/prompt` with a
JSON-RPC error
FAKE_ACP_PERMISSION_AFTER_CANCEL=1 defer the first prompt's permission
callback until a later prompt
Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick <n>",
counting from 1) and logs `emit tick <n>`, so a test can make the agent
@@ -70,6 +73,9 @@ signal.signal(signal.SIGTERM, on_sigterm)
session_id = f"fake-{os.getpid()}"
ticks = 0
prompt_failures = 0
pending_prompt_id = None
pending_permission_id = None
delayed_permission = False
def chunk(text):
@@ -102,6 +108,12 @@ for raw in sys.stdin:
method = msg.get("method")
if method is None:
# A response to something we sent; nothing to do.
if msg.get("id") == pending_permission_id:
log("permission response " + json.dumps(msg.get("result"), sort_keys=True))
send({"jsonrpc": "2.0", "id": pending_prompt_id,
"result": {"stopReason": "end_turn"}})
pending_prompt_id = None
pending_permission_id = None
continue
log(f"recv {method}")
rid = msg.get("id")
@@ -140,9 +152,47 @@ for raw in sys.stdin:
elif method == "session/cancel":
if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
send(chunk("final on cancel"))
if (os.environ.get("FAKE_ACP_CANCEL_SETTLES") == "1"
and pending_prompt_id is not None):
send({"jsonrpc": "2.0", "id": pending_prompt_id,
"result": {"stopReason": "cancelled"}})
pending_prompt_id = None
pending_permission_id = None
if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
sys.exit(0)
elif method == "session/prompt":
if os.environ.get("FAKE_ACP_PERMISSION_AFTER_CANCEL") == "1":
if pending_prompt_id is None:
pending_prompt_id = rid
delayed_permission = True
continue
if delayed_permission:
delayed_permission = False
pending_permission_id = "permission-1"
send({"jsonrpc": "2.0", "id": pending_permission_id,
"method": "session/request_permission", "params": {
"sessionId": session_id,
"toolCall": {"toolCallId": "tool-1", "fields": {}},
"options": [
{"optionId": "once", "name": "Allow once", "kind": "allow_once"},
{"optionId": "deny", "name": "Deny", "kind": "reject_once"}
]
}})
continue
if os.environ.get("FAKE_ACP_REQUEST_PERMISSION") == "1":
pending_prompt_id = rid
pending_permission_id = "permission-1"
send({"jsonrpc": "2.0", "id": pending_permission_id,
"method": "session/request_permission", "params": {
"sessionId": session_id,
"toolCall": {"toolCallId": "tool-1", "fields": {}},
"options": [
{"optionId": "once", "name": "Allow once", "kind": "allow_once"},
{"optionId": "session", "name": "Allow session", "kind": "allow_always"},
{"optionId": "deny", "name": "Deny", "kind": "reject_once"}
]
}})
continue
time.sleep(int(os.environ.get("FAKE_ACP_DELAY_PROMPT_MS", "0")) / 1000)
fail_once = os.environ.get("FAKE_ACP_PROMPT_FAIL_ONCE") == "1"
if os.environ.get("FAKE_ACP_PROMPT_FAIL") == "1" or (fail_once and prompt_failures == 0):
+48 -40
View File
@@ -21,13 +21,12 @@ use maestro_config::SpawnPreset;
use maestro_protocol::{
routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
StructuredApprovalDecision, StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
};
use serde::Deserialize;
use uuid::Uuid;
use crate::projection::TurnLifecycle;
use crate::session::SessionManager;
use crate::session::{QueueAcpPrompt, SessionManager};
use crate::sidecar::EventCursor;
#[derive(Debug, Clone)]
@@ -70,7 +69,11 @@ pub fn router(state: LocalApiState) -> Router {
.route(routes::patterns::STRUCTURED, any(structured_ws))
.route(routes::patterns::STRUCTURED_REPLAY, get(structured_replay))
.route(routes::patterns::STRUCTURED_PROMPT, post(structured_prompt))
.route(routes::patterns::STRUCTURED_CANCEL, post(structured_cancel));
.route(routes::patterns::STRUCTURED_CANCEL, post(structured_cancel))
.route(
routes::patterns::STRUCTURED_APPROVAL,
post(structured_approval),
);
#[cfg(feature = "haiku-autoname")]
{
router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
@@ -432,45 +435,24 @@ async fn structured_prompt(
Path(id): Path<Uuid>,
Json(body): Json<StructuredPromptRequest>,
) -> Response {
let (acp, turn) = match s.manager.begin_acp_turn(id).await {
Ok(crate::session::BeginAcpTurn::Started(acp, turn)) => (acp, turn),
Ok(crate::session::BeginAcpTurn::InFlight) => {
return error_response(
StatusCode::CONFLICT,
"turn_in_flight",
"the structured session already has an active turn",
)
}
Ok(crate::session::BeginAcpTurn::Unavailable) => {
return error_response(
StatusCode::CONFLICT,
"not_structured",
"session is not a live structured session",
)
match s.manager.queue_acp_prompt(id, body.prompt).await {
Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(),
Ok(QueueAcpPrompt::InFlight) => error_response(
StatusCode::CONFLICT,
"turn_in_flight",
"the structured session already has an active turn",
),
Ok(QueueAcpPrompt::Unavailable) => error_response(
StatusCode::CONFLICT,
"not_structured",
"session is not a live structured session",
),
Ok(QueueAcpPrompt::Failed(message)) => {
error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
}
Err(e) => {
let info = e.to_error_info();
return error_response(session_error_status(&info.code), &info.code, &info.message);
}
};
match acp.prompt(body.prompt) {
Ok(done) => {
let manager = s.manager.clone();
tokio::spawn(async move {
let lifecycle = match done.await {
Ok(Ok(())) => TurnLifecycle::Finish,
Ok(Err(_)) | Err(_) => TurnLifecycle::Failure,
};
let _ = manager.apply_acp_turn_lifecycle(id, lifecycle, turn).await;
});
StatusCode::NO_CONTENT.into_response()
}
Err(message) => {
let _ = s
.manager
.apply_acp_turn_lifecycle(id, TurnLifecycle::Failure, turn)
.await;
error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
error_response(session_error_status(&info.code), &info.code, &info.message)
}
}
}
@@ -490,6 +472,32 @@ async fn structured_cancel(State(s): State<LocalApiState>, Path(id): Path<Uuid>)
}
}
async fn structured_approval(
State(s): State<LocalApiState>,
Path((id, nonce)): Path<(Uuid, String)>,
Json(decision): Json<StructuredApprovalDecision>,
) -> Response {
let decision = match decision {
StructuredApprovalDecision::AllowOnce => crate::session::ApprovalDecision::AllowOnce,
StructuredApprovalDecision::AllowForSession => {
crate::session::ApprovalDecision::AllowForSession
}
StructuredApprovalDecision::Deny => crate::session::ApprovalDecision::Deny,
};
match s.manager.resolve_acp_approval(id, &nonce, decision).await {
Ok(Some(())) => StatusCode::NO_CONTENT.into_response(),
Ok(None) => error_response(
StatusCode::CONFLICT,
"not_structured",
"session is not a live structured session",
),
Err(e) => {
let info = e.to_error_info();
error_response(session_error_status(&info.code), &info.code, &info.message)
}
}
}
/// Convenience constructor — pulls the bits `LocalApiState` needs from
/// `DaemonConfig` so callers don't have to assemble it by hand. The project
/// scanner starts with its cache cold; wire
+119
View File
@@ -155,6 +155,61 @@ impl Projection {
self.active_turn
}
/// Add a daemon-originated permission row. The nonce belongs only in the
/// reduced state, which is the client control surface, never in ACP data.
pub fn request_approval(&mut self, nonce: String, request: Value) -> ProjectionUpdate {
let row = self.append(
format!("approval:{}", uuid::Uuid::new_v4()),
TranscriptRowKind::Approval,
request.clone(),
);
self.state.pending_approval = Some(serde_json::json!({
"nonce": nonce,
"request": request,
}));
ProjectionUpdate {
transcript: Some(row),
state: self.state(),
}
}
pub fn restore_approval(&mut self, row: TranscriptRow) -> ProjectionUpdate {
let transcript = if !self.row_index.contains_key(&row.id) {
self.row_index.insert(row.id.clone(), self.rows.len());
self.rows.push(row.clone());
Some(TranscriptDelta::Append { row })
} else {
None
};
ProjectionUpdate {
transcript,
state: self.state(),
}
}
/// Clear an approval only when it is still the one the client answered.
pub fn resolve_approval(&mut self, nonce: &str) -> Option<ProjectionUpdate> {
let pending = self.state.pending_approval.as_ref()?;
if pending.get("nonce").and_then(Value::as_str) != Some(nonce) {
return None;
}
self.state.pending_approval = None;
Some(ProjectionUpdate {
transcript: None,
state: self.state(),
})
}
/// Terminal turn paths must not leave a recovered snapshot permanently
/// blocked on an ACP request whose transport no longer exists.
pub fn clear_approval(&mut self) -> Option<ProjectionUpdate> {
self.state.pending_approval.take()?;
Some(ProjectionUpdate {
transcript: None,
state: self.state(),
})
}
fn content(
&mut self,
seq: u64,
@@ -632,4 +687,68 @@ mod tests {
fn failed_turn_does_not_reactivate_unfinished_tools() {
terminal_turn_drops_unfinished_tools(TurnLifecycle::Failure, "failed-1", "next-1");
}
#[test]
fn approval_history_survives_pending_cleanup() {
let mut projection = Projection::default();
let update =
projection.request_approval("secret".into(), serde_json::json!({"tool":"bash"}));
let TranscriptDelta::Append { row } = update.transcript.unwrap() else {
panic!("approval row")
};
assert_eq!(row.kind, TranscriptRowKind::Approval);
assert!(row.data.get("nonce").is_none());
assert!(!row.id.contains("secret"));
let persisted =
serde_json::to_string(&crate::sidecar::PersistedEvent::Approval { row: row.clone() })
.unwrap();
assert!(!persisted.contains("secret"));
assert_eq!(
projection.state().pending_approval.unwrap()["nonce"],
"secret"
);
projection.clear_approval();
assert!(projection.state().pending_approval.is_none());
assert_eq!(projection.snapshot(0).rows, vec![row]);
}
#[test]
fn recovered_approval_is_historical_not_pending() {
let mut live = Projection::default();
let TranscriptDelta::Append { row } = live
.request_approval("daemon-nonce".into(), serde_json::json!({"tool":"bash"}))
.transcript
.unwrap()
else {
panic!("approval row")
};
let mut recovered = Projection::default();
let update = recovered.restore_approval(row.clone());
assert_eq!(
update.transcript,
Some(TranscriptDelta::Append { row: row.clone() })
);
assert!(update.state.pending_approval.is_none());
assert_eq!(recovered.snapshot(0).rows, vec![row]);
let replay = serde_json::to_string(&recovered.snapshot(1).rows).unwrap();
assert!(!replay.contains("daemon-nonce"));
}
#[test]
fn stale_terminal_turn_cannot_clear_newer_approval() {
let mut projection = Projection::default();
let (first, _) = projection.begin_turn();
projection.turn_lifecycle(TurnLifecycle::Cancel, first);
let (second, _) = projection.begin_turn();
projection.request_approval("second-nonce".into(), serde_json::json!({"tool":"bash"}));
assert!(projection
.turn_lifecycle(TurnLifecycle::Finish, first)
.is_none());
assert_eq!(projection.active_turn(), Some(second));
assert_eq!(
projection.state().pending_approval.unwrap()["nonce"],
"second-nonce"
);
}
}
+464 -14
View File
@@ -12,18 +12,21 @@ use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use maestro_acp::{AcpSession, SpawnConfig};
use maestro_acp::{AcpSession, PermissionRequest, SpawnConfig};
use maestro_protocol::{
ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
SessionStatus, SessionView, StructuredReplayEvent, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, WorktreeSpec, NAME_KEY,
NAME_SOURCE_KEY,
};
use tokio::sync::{broadcast, watch, Mutex};
use tokio::sync::{broadcast, oneshot, watch, Mutex};
use uuid::Uuid;
use crate::panes::Panes;
use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
use crate::pty::{pipe_pane_command, PtyBridge};
use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore};
use crate::sidecar::{
EventCursor, EventLog, LoggedEvent, PersistedEvent, SidecarError, SidecarStore,
};
use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
use crate::tmux::TmuxError;
use crate::viewport::{ViewportId, Viewports};
@@ -118,6 +121,33 @@ struct AcpEventStream {
live: broadcast::Sender<LoggedEvent>,
projection: Projection,
projected_live: broadcast::Sender<ProjectionUpdate>,
pending_approvals: HashMap<String, PendingApproval>,
/// Serializes prompt dispatch with cancellation. A prompt is not merely
/// claimed: it is queued onto ACP before another operation can proceed.
operation: Arc<Mutex<()>>,
dispatch: AcpDispatchState,
}
#[derive(Debug, Default)]
struct AcpDispatchState {
queued_turn: Option<u64>,
cancelled_turn: Option<u64>,
failed: bool,
settled: Arc<tokio::sync::Notify>,
}
#[derive(Debug)]
struct PendingApproval {
response: oneshot::Sender<maestro_acp::schema::RequestPermissionOutcome>,
options: Vec<maestro_acp::schema::PermissionOption>,
}
/// The outcome accepted by the structured approval endpoint.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalDecision {
AllowOnce,
AllowForSession,
Deny,
}
#[derive(Debug, thiserror::Error)]
@@ -156,6 +186,15 @@ pub enum BeginAcpTurn {
Unavailable,
}
/// Result of serializing a projected turn claim with its ACP prompt enqueue.
#[derive(Debug)]
pub enum QueueAcpPrompt {
Queued,
InFlight,
Unavailable,
Failed(String),
}
impl SessionError {
pub fn to_error_info(&self) -> ErrorInfo {
let code = match self {
@@ -406,7 +445,7 @@ impl SessionManager {
let mut projection = Projection::default();
let mut replay = Vec::new();
for event in events {
let update = projection.apply(event.seq, &event.notification);
let update = apply_persisted_event(&mut projection, &event);
let include = match cursor {
EventCursor::Since(since) => event.seq > since,
EventCursor::Before(Some(before)) => event.seq < before,
@@ -488,6 +527,10 @@ impl SessionManager {
return Ok(false);
};
let mut stream = stream.lock().await;
if stream.projection.active_turn() != Some(turn) {
return Ok(false);
}
cancel_pending_approvals(&mut stream);
let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) else {
return Ok(false);
};
@@ -518,6 +561,150 @@ impl SessionManager {
Ok(BeginAcpTurn::Started(acp, turn))
}
/// Claim a turn and enqueue its ACP prompt as one serialized operation.
///
/// ACP permission callbacks do not carry a prompt id. Therefore a
/// cancelled prompt must settle before a later prompt is dispatched. If
/// it does not settle within the cancel grace, the session is failed
/// instead of letting a later callback become ambiguous.
pub async fn queue_acp_prompt(
&self,
id: Uuid,
text: String,
) -> Result<QueueAcpPrompt, SessionError> {
let (acp, stream) = {
let sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if !record.is_live_structured() {
return Ok(QueueAcpPrompt::Unavailable);
}
(record.acp.clone(), record.acp_events.clone())
};
let (Some(acp), Some(stream)) = (acp, stream) else {
return Ok(QueueAcpPrompt::Unavailable);
};
let operation = stream.lock().await.operation.clone();
let _operation = operation.lock().await;
loop {
let waiting_for = {
let mut stream = stream.lock().await;
if stream.dispatch.failed {
return Ok(QueueAcpPrompt::Unavailable);
}
if let Some(turn) = stream.dispatch.queued_turn {
if stream.dispatch.cancelled_turn != Some(turn) {
return Ok(QueueAcpPrompt::InFlight);
}
Some((turn, stream.dispatch.settled.clone()))
} else {
if stream.projection.turn_active() {
return Ok(QueueAcpPrompt::InFlight);
}
let (turn, update) = stream.projection.begin_turn();
let _ = stream.projected_live.send(update);
match acp.prompt(text.clone(), turn) {
Ok(done) => {
stream.dispatch.queued_turn = Some(turn);
let manager = self.clone();
tokio::spawn(async move {
let lifecycle = match done.await {
Ok(Ok(())) => TurnLifecycle::Finish,
Ok(Err(_)) | Err(_) => TurnLifecycle::Failure,
};
manager.finish_queued_acp_turn(id, turn, lifecycle).await;
});
return Ok(QueueAcpPrompt::Queued);
}
Err(message) => {
cancel_pending_approvals(&mut stream);
let projected = stream
.projection
.turn_lifecycle(TurnLifecycle::Failure, turn)
.expect(
"newly claimed turn remains current while stream is locked",
);
let _ = stream.projected_live.send(projected);
return Ok(QueueAcpPrompt::Failed(message));
}
}
}
};
let Some((turn, settled)) = waiting_for else {
continue;
};
if tokio::time::timeout(maestro_acp::CANCEL_GRACE, settled.notified())
.await
.is_err()
{
self.fail_cancelled_acp_turn(id, &stream, turn).await;
}
}
}
async fn finish_queued_acp_turn(&self, id: Uuid, turn: u64, lifecycle: TurnLifecycle) {
let stream = self
.inner
.sessions
.lock()
.await
.get(&id)
.and_then(|record| record.acp_events.clone());
let Some(stream) = stream else {
return;
};
let mut stream = stream.lock().await;
if stream.dispatch.queued_turn != Some(turn) {
return;
}
stream.dispatch.queued_turn = None;
if stream.dispatch.cancelled_turn == Some(turn) {
stream.dispatch.cancelled_turn = None;
}
stream.dispatch.settled.notify_waiters();
if stream.projection.active_turn() != Some(turn) {
return;
}
cancel_pending_approvals(&mut stream);
if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
let _ = stream.projected_live.send(projected);
}
}
async fn fail_cancelled_acp_turn(
&self,
id: Uuid,
stream: &Arc<Mutex<AcpEventStream>>,
turn: u64,
) {
let acp = self
.inner
.sessions
.lock()
.await
.get(&id)
.and_then(|record| record.acp.clone());
{
let mut stream = stream.lock().await;
if stream.dispatch.queued_turn != Some(turn)
|| stream.dispatch.cancelled_turn != Some(turn)
{
return;
}
stream.dispatch.failed = true;
stream.dispatch.queued_turn = None;
stream.dispatch.cancelled_turn = None;
stream.dispatch.settled.notify_waiters();
}
mark_session_dead(&self.inner, id).await;
if let Some(acp) = acp {
tokio::spawn(async move {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
});
}
}
/// Cancel an ACP turn and immediately clear its client-facing lifecycle.
pub async fn cancel_acp_turn(&self, id: Uuid) -> Result<bool, SessionError> {
let (acp, stream) = {
@@ -531,19 +718,93 @@ impl SessionManager {
let (Some(acp), Some(stream)) = (acp, stream) else {
return Ok(false);
};
let mut stream = stream.lock().await;
let event_stream = stream;
let operation = event_stream.lock().await.operation.clone();
let _operation = operation.lock().await;
let mut stream = event_stream.lock().await;
acp.cancel();
cancel_pending_approvals(&mut stream);
let cancelled_queued_turn = stream.dispatch.queued_turn;
if let Some(turn) = cancelled_queued_turn {
stream.dispatch.cancelled_turn = Some(turn);
}
let Some(turn) = stream.projection.active_turn() else {
drop(stream);
if let Some(turn) = cancelled_queued_turn {
self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
}
return Ok(true);
};
acp.cancel();
let projected = stream
.projection
.turn_lifecycle(TurnLifecycle::Cancel, turn)
.expect("active turn remains current while stream is locked");
let _ = stream.projected_live.send(projected);
drop(stream);
if let Some(turn) = cancelled_queued_turn {
self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
}
Ok(true)
}
fn schedule_cancelled_acp_turn_failure(
&self,
id: Uuid,
stream: Arc<Mutex<AcpEventStream>>,
turn: u64,
) {
let manager = self.clone();
tokio::spawn(async move {
tokio::time::sleep(maestro_acp::CANCEL_GRACE).await;
manager.fail_cancelled_acp_turn(id, &stream, turn).await;
});
}
/// Resolve a daemon-issued approval nonce. Missing or already-consumed
/// nonces deliberately succeed: two clients racing to answer must not
/// turn a harmless duplicate click into an error.
pub async fn resolve_acp_approval(
&self,
id: Uuid,
nonce: &str,
decision: ApprovalDecision,
) -> Result<Option<()>, SessionError> {
let stream = {
let sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if !record.is_live_structured() {
return Ok(None);
}
record.acp_events.clone()
};
let Some(stream) = stream else {
return Ok(None);
};
let mut stream = stream.lock().await;
let Some(pending) = stream.pending_approvals.remove(nonce) else {
return Ok(Some(()));
};
let outcome = permission_outcome(decision, &pending.options);
let _ = pending.response.send(outcome);
if let Some(update) = stream.projection.resolve_approval(nonce) {
let _ = stream.projected_live.send(update);
}
Ok(Some(()))
}
async fn cancel_acp_approvals(&self, id: Uuid) {
let stream = self
.inner
.sessions
.lock()
.await
.get(&id)
.and_then(|record| record.acp_events.clone());
if let Some(stream) = stream {
cancel_pending_approvals(&mut *stream.lock().await);
}
}
/// End every adapter child before the daemon exits (#134): each gets
/// `session/cancel`, then SIGTERM, concurrently. The sessions stay in
/// the map, marked Dead and written through to their sidecars, so what
@@ -569,6 +830,9 @@ impl SessionManager {
if children.is_empty() {
return;
}
for (id, _, _) in &children {
self.cancel_acp_approvals(*id).await;
}
let reaped =
futures_util::future::join_all(children.iter().map(|(_, acp, _)| acp.reap_if_exited()))
.await;
@@ -818,13 +1082,16 @@ impl SessionManager {
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
let mut projection = Projection::default();
for event in &persisted {
projection.apply(event.seq, &event.notification);
apply_persisted_event(&mut projection, event);
}
Some(Arc::new(Mutex::new(AcpEventStream {
log,
live: tx,
projection,
projected_live: projected_tx,
pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(),
})))
}
Err(e) => {
@@ -933,7 +1200,8 @@ impl SessionManager {
);
}
if let Some((acp, events)) = event_pump {
let done = start_acp_event_pump(&self.inner, id, &acp, events);
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&acp, events);
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
record.acp_event_pump = Some(done);
}
@@ -1260,13 +1528,16 @@ impl SessionManager {
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
let mut projection = Projection::default();
for event in &persisted {
projection.apply(event.seq, &event.notification);
apply_persisted_event(&mut projection, event);
}
Some(Arc::new(Mutex::new(AcpEventStream {
log,
live: tx,
projection,
projected_live: projected_tx,
pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(),
})))
}
Err(error) => {
@@ -1318,7 +1589,8 @@ impl SessionManager {
},
);
if let Some((acp, events)) = event_pump {
let done = start_acp_event_pump(&self.inner, id, &acp, events);
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&acp, events);
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
record.acp_event_pump = Some(done);
}
@@ -1672,10 +1944,15 @@ fn start_acp_event_pump(
while let Some(notification) = updates.recv().await {
let append_error = {
let mut events = events.lock().await;
match events.log.append(&notification) {
let event = PersistedEvent::Notification(Box::new(notification));
match events.log.append(&event) {
Ok(seq) => {
let projected = events.projection.apply(seq, &notification);
let _ = events.live.send(LoggedEvent { seq, notification });
let logged = LoggedEvent {
seq,
event: event.clone(),
};
let projected = apply_persisted_event(&mut events.projection, &logged);
let _ = events.live.send(logged);
let _ = events.projected_live.send(projected);
None
}
@@ -1699,6 +1976,112 @@ fn start_acp_event_pump(
done_rx
}
fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> ProjectionUpdate {
match &event.event {
PersistedEvent::Notification(notification) => projection.apply(event.seq, notification),
PersistedEvent::Approval { row } => projection.restore_approval(row.clone()),
}
}
fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventStream>>) {
let Some(mut requests) = acp.take_permission_requests() else {
return;
};
tokio::spawn(async move {
while let Some(PermissionRequest {
turn,
request,
response,
}) = requests.recv().await
{
let mut events = events.lock().await;
if events.projection.active_turn() != Some(turn) {
let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
continue;
}
let nonce = Uuid::new_v4().to_string();
let data = match serde_json::to_value(&request) {
Ok(value) => value,
Err(_) => {
let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
continue;
}
};
// ACP requests serialize through its connection. Keep the newest
// one authoritative if an adapter nonetheless overlaps them.
cancel_pending_approvals(&mut events);
events.pending_approvals.insert(
nonce.clone(),
PendingApproval {
response,
options: request.options,
},
);
let update = events.projection.request_approval(nonce, data);
if let Some(TranscriptDelta::Append { row }) = &update.transcript {
let event = PersistedEvent::Approval { row: row.clone() };
if let Err(error) = events.log.append(&event) {
tracing::error!(%error, "approval history append failed");
cancel_pending_approvals(&mut events);
continue;
}
let seq = events.log.last_seq();
let _ = events.live.send(LoggedEvent { seq, event });
}
let _ = events.projected_live.send(update);
}
});
}
fn cancel_pending_approvals(events: &mut AcpEventStream) {
for (_, pending) in events.pending_approvals.drain() {
let _ = pending
.response
.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
}
if let Some(update) = events.projection.clear_approval() {
let _ = events.projected_live.send(update);
}
}
fn permission_outcome(
decision: ApprovalDecision,
options: &[maestro_acp::schema::PermissionOption],
) -> maestro_acp::schema::RequestPermissionOutcome {
use maestro_acp::schema::{
PermissionOptionKind, RequestPermissionOutcome, SelectedPermissionOutcome,
};
let kind = match decision {
ApprovalDecision::AllowOnce => PermissionOptionKind::AllowOnce,
ApprovalDecision::AllowForSession => PermissionOptionKind::AllowAlways,
ApprovalDecision::Deny => {
// Prefer a one-shot refusal; use a permanent refusal only when
// that is the sole rejection option the ACP request offered.
return options
.iter()
.find(|option| option.kind == PermissionOptionKind::RejectOnce)
.or_else(|| {
options
.iter()
.find(|option| option.kind == PermissionOptionKind::RejectAlways)
})
.map_or(RequestPermissionOutcome::Cancelled, |option| {
RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
option.option_id.clone(),
))
});
}
};
options.iter().find(|option| option.kind == kind).map_or(
RequestPermissionOutcome::Cancelled,
|option| {
RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
option.option_id.clone(),
))
},
)
}
async fn wait_for_event_pump(mut done: watch::Receiver<bool>) {
if !*done.borrow() {
let _ = done.wait_for(|finished| *finished).await;
@@ -2024,6 +2407,15 @@ async fn mark_session_dead_with_worktree_settlement(
let Some((mut session, worktree)) = flipped else {
return false;
};
let stream = inner
.sessions
.lock()
.await
.get(&id)
.and_then(|record| record.acp_events.clone());
if let Some(stream) = stream {
cancel_pending_approvals(&mut *stream.lock().await);
}
tracing::info!(%id, "session process gone — marking dead");
if settle_worktree {
@@ -2130,6 +2522,9 @@ async fn cleanup_record(
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
}
if let Some(acp) = &record.acp {
if let Some(stream) = &record.acp_events {
cancel_pending_approvals(&mut *stream.lock().await);
}
// `session/cancel`, a short wait, then SIGTERM (#134).
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
@@ -2163,3 +2558,58 @@ async fn cleanup_record(
}
let _ = inner.events.send(CatalogEvent::SessionRemoved { id });
}
#[cfg(test)]
mod tests {
use super::{permission_outcome, ApprovalDecision};
use maestro_acp::schema::{PermissionOption, PermissionOptionKind, RequestPermissionOutcome};
fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
PermissionOption::new(id.to_owned(), id, kind)
}
fn selected_id(outcome: RequestPermissionOutcome) -> Option<String> {
match outcome {
RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()),
RequestPermissionOutcome::Cancelled => None,
_ => None,
}
}
#[test]
fn permission_decisions_select_only_offered_options() {
let all = [
option("allow-once", PermissionOptionKind::AllowOnce),
option("allow-always", PermissionOptionKind::AllowAlways),
option("reject-once", PermissionOptionKind::RejectOnce),
option("reject-always", PermissionOptionKind::RejectAlways),
];
assert_eq!(
selected_id(permission_outcome(ApprovalDecision::AllowOnce, &all)).as_deref(),
Some("allow-once")
);
assert_eq!(
selected_id(permission_outcome(ApprovalDecision::AllowForSession, &all)).as_deref(),
Some("allow-always")
);
assert_eq!(
selected_id(permission_outcome(ApprovalDecision::Deny, &all)).as_deref(),
Some("reject-once")
);
assert_eq!(
selected_id(permission_outcome(
ApprovalDecision::Deny,
&[option("reject-always", PermissionOptionKind::RejectAlways)],
))
.as_deref(),
Some("reject-always")
);
assert_eq!(
selected_id(permission_outcome(
ApprovalDecision::Deny,
&[option("allow-once", PermissionOptionKind::AllowOnce)],
)),
None
);
}
}
+55 -9
View File
@@ -7,8 +7,8 @@
//! advertised but with empty metadata (handled by the caller, not here).
//!
//! A structured session (#135) also owns `<id>.events.jsonl` next to its
//! sidecar: every `session/update` notification its adapter sent, one JSON
//! object per line, appended before it is broadcast. The log is the source of
//! sidecar: every `session/update` notification and daemon approval row is one
//! JSON object, appended before it is broadcast. The log is the source of
//! truth for the transcript; `seq` is the line's 1-based position, so a
//! client that holds seq `n` asks for [`EventCursor::Since`]`(n)`. A line
//! that does not parse — including the partial line a hard kill leaves at
@@ -21,7 +21,7 @@ use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use maestro_acp::schema::SessionNotification;
use maestro_protocol::Session;
use maestro_protocol::{Session, TranscriptRow};
use uuid::Uuid;
#[derive(Debug, thiserror::Error)]
@@ -150,10 +150,10 @@ impl SidecarStore {
}
continue;
}
let Ok(notification) = serde_json::from_slice::<SessionNotification>(&line) else {
let Ok(event) = PersistedEvent::decode(&line) else {
continue;
};
out.push_back(LoggedEvent { seq, notification });
out.push_back(LoggedEvent { seq, event });
match cursor {
EventCursor::Since(_) if out.len() >= limit => break,
EventCursor::Before(_) if out.len() > limit => {
@@ -190,6 +190,7 @@ impl SidecarStore {
/// whichever does not exist.
pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
remove_if_present(&self.events_path_for(id))?;
remove_if_present(&self.dir.join(format!("{id}.approvals.jsonl")))?;
remove_if_present(&self.recovery_path_for(id))?;
remove_if_present(&self.path_for(id))
}
@@ -247,11 +248,52 @@ pub enum EventCursor {
}
/// One line of a session's event log.
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[derive(Debug, Clone, PartialEq)]
pub struct LoggedEvent {
/// 1-based line position in the log; the cursor for [`EventCursor`].
pub seq: u64,
pub notification: SessionNotification,
pub event: PersistedEvent,
}
/// A single entry in the structured event history.
///
/// ACP notifications retain their original JSON representation, so existing
/// event logs remain readable. Daemon-generated rows use an envelope so they
/// can share the same cursor sequence without pretending to be ACP traffic.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "maestroEvent", rename_all = "snake_case")]
pub enum PersistedEvent {
Approval {
row: TranscriptRow,
},
#[serde(skip)]
Notification(Box<SessionNotification>),
}
impl PersistedEvent {
fn decode(line: &[u8]) -> Result<Self, serde_json::Error> {
serde_json::from_slice::<SessionNotification>(line)
.map(|notification| Self::Notification(Box::new(notification)))
.or_else(|_| serde_json::from_slice(line))
}
}
impl From<SessionNotification> for PersistedEvent {
fn from(notification: SessionNotification) -> Self {
Self::Notification(Box::new(notification))
}
}
impl From<&SessionNotification> for PersistedEvent {
fn from(notification: &SessionNotification) -> Self {
Self::Notification(Box::new(notification.clone()))
}
}
impl From<&PersistedEvent> for PersistedEvent {
fn from(event: &PersistedEvent) -> Self {
event.clone()
}
}
/// An open, append-only handle on one session's event log. Obtained from
@@ -272,8 +314,12 @@ impl EventLog {
/// handed to the kernel in a single write before this returns, so a
/// caller that broadcasts afterwards never announces an event the log
/// would not replay — short of the machine losing power.
pub fn append(&mut self, notification: &SessionNotification) -> Result<u64, SidecarError> {
let mut line = serde_json::to_vec(notification)?;
pub fn append(&mut self, event: impl Into<PersistedEvent>) -> Result<u64, SidecarError> {
let event = event.into();
let mut line = match &event {
PersistedEvent::Notification(notification) => serde_json::to_vec(notification)?,
_ => serde_json::to_vec(&event)?,
};
line.push(b'\n');
self.file.write_all(&line)?;
self.file.flush()?;
+105 -1
View File
@@ -83,7 +83,10 @@ impl Fake {
}
fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
match &event.notification.update {
let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
panic!("unexpected daemon event")
};
match &notification.update {
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
ContentBlock::Text(text) => &text.text,
other => panic!("unexpected content {other:?}"),
@@ -321,6 +324,107 @@ async fn stale_turn_completion_cannot_end_the_next_turn() {
fx.manager.kill(session.id, false).await.unwrap();
}
#[tokio::test]
async fn cancelled_prompt_that_never_settles_fails_the_session() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_PERMISSION_AFTER_CANCEL".into(), "1".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.unwrap();
assert!(matches!(
fx.manager
.queue_acp_prompt(session.id, "first".into())
.await
.unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued
));
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
let second = fx
.manager
.queue_acp_prompt(session.id, "second".into())
.await
.unwrap();
assert!(
matches!(second, maestro_daemon::session::QueueAcpPrompt::Unavailable),
"{second:?}"
);
assert_eq!(
fx.manager.status(session.id).await,
Some(SessionStatus::Dead)
);
assert!(!fake.log().contains("permission response"));
}
#[tokio::test]
async fn cancellation_settlement_allows_the_next_prompt_permission() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
env.insert("FAKE_ACP_CANCEL_SETTLES".into(), "1".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.unwrap();
assert!(matches!(
fx.manager
.queue_acp_prompt(session.id, "first".into())
.await
.unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued
));
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
assert!(matches!(
fx.manager
.queue_acp_prompt(session.id, "second".into())
.await
.unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued
));
let nonce = tokio::time::timeout(Duration::from_secs(2), async {
loop {
let (_, state, _) = fx
.manager
.subscribe_acp_projection(session.id, EventCursor::Since(0))
.await
.unwrap()
.unwrap();
if let Some(pending) = state.pending_approval {
break pending["nonce"].as_str().unwrap().to_owned();
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.unwrap();
assert!(fx
.manager
.resolve_acp_approval(
session.id,
&nonce,
maestro_daemon::session::ApprovalDecision::AllowOnce,
)
.await
.unwrap()
.is_some());
wait_for_async(Duration::from_secs(2), || {
let log = fake.log();
async move { log.contains("permission response") }
})
.await;
fx.manager.kill(session.id, false).await.unwrap();
}
#[tokio::test]
async fn recovery_loads_the_acp_session_without_duplicating_history() {
let fx = DaemonFixture::new();
+133
View File
@@ -142,6 +142,10 @@ impl FakeAcp {
.next_back()
.expect("fake ACP started")
}
fn log(&self) -> String {
std::fs::read_to_string(self.dir.path().join("agent.log")).unwrap_or_default()
}
}
async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid {
@@ -256,6 +260,135 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it(
.await
.unwrap();
assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT);
tokio::time::timeout(Duration::from_secs(2), async {
loop {
if fake.log().matches("recv session/prompt").count() == 2 {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("both prompt commands reach ACP");
let log = fake.log();
let first_prompt = log.find("recv session/prompt").unwrap();
let cancel = log.find("recv session/cancel").unwrap();
let second_prompt = log.rfind("recv session/prompt").unwrap();
assert!(first_prompt < cancel && cancel < second_prompt, "{log}");
}
#[tokio::test]
async fn structured_approval_is_secret_single_use_and_race_safe() {
let h = spawn("naps-laptop").await;
let fake = FakeAcp::new();
let mut env = fake.env();
env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
let id = structured_session_with_env(&h, env).await;
let client = reqwest::Client::new();
let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
.await
.unwrap();
ws.next().await.unwrap().unwrap();
ws.next().await.unwrap().unwrap();
assert_eq!(
client
.post(url(
h.addr,
&format!("/api/sessions/{id}/structured/prompt")
))
.json(&serde_json::json!({"prompt":"needs permission"}))
.send()
.await
.unwrap()
.status(),
reqwest::StatusCode::NO_CONTENT
);
let nonce = tokio::time::timeout(Duration::from_secs(2), async {
loop {
let event: serde_json::Value =
serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
.unwrap();
if let Some(nonce) = event["state"]["pending_approval"]["nonce"].as_str() {
assert_eq!(event["transcript"]["row"]["kind"], "approval");
assert!(event["transcript"]["row"]["data"].get("nonce").is_none());
break nonce.to_owned();
}
}
})
.await
.unwrap();
assert!(
!fake.log().contains(&nonce),
"daemon nonce leaked to ACP agent"
);
let path = format!("/api/sessions/{id}/structured/approvals/{nonce}");
let (a, b) = tokio::join!(
client
.post(url(h.addr, &path))
.json(&serde_json::json!("allow_once"))
.send(),
client
.post(url(h.addr, &path))
.json(&serde_json::json!("deny"))
.send(),
);
assert_eq!(a.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
assert_eq!(b.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
assert_eq!(
client
.post(url(h.addr, &path))
.json(&serde_json::json!("deny"))
.send()
.await
.unwrap()
.status(),
reqwest::StatusCode::NO_CONTENT
);
tokio::time::timeout(Duration::from_secs(2), async {
loop {
if fake.log().contains("permission response") {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.unwrap();
let log = fake.log();
assert!(
log.contains("\"optionId\": \"once\"") || log.contains("\"optionId\": \"deny\""),
"{log}"
);
tokio::time::timeout(Duration::from_secs(1), async {
loop {
let event: serde_json::Value =
serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
.unwrap();
if event["state"]["pending_approval"].is_null() {
break;
}
}
})
.await
.expect("resolved approval clears the reduced state");
assert_eq!(
client
.post(url(
h.addr,
&format!(
"/api/sessions/{}/structured/approvals/{nonce}",
Uuid::new_v4()
)
))
.json(&serde_json::json!("deny"))
.send()
.await
.unwrap()
.status(),
reqwest::StatusCode::NOT_FOUND
);
}
#[tokio::test]
+64 -16
View File
@@ -6,8 +6,8 @@ use std::collections::HashMap;
use std::path::PathBuf;
use chrono::{DateTime, Utc};
use maestro_daemon::sidecar::SidecarStore;
use maestro_protocol::{Session, SessionStatus, SessionView};
use maestro_daemon::sidecar::{PersistedEvent, SidecarStore};
use maestro_protocol::{Session, SessionStatus, SessionView, TranscriptRow, TranscriptRowKind};
use serde_json::Value;
use tempfile::TempDir;
use uuid::Uuid;
@@ -186,6 +186,13 @@ fn text_of(n: &SessionNotification) -> &str {
}
}
fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
panic!("expected ACP notification")
};
text_of(notification)
}
#[test]
fn append_assigns_line_position_as_seq_and_reads_back() {
let dir = TempDir::new().unwrap();
@@ -194,9 +201,9 @@ fn append_assigns_line_position_as_seq_and_reads_back() {
let mut log = store.open_events(id).expect("open");
assert_eq!(log.last_seq(), 0);
assert_eq!(log.append(&note("a")).unwrap(), 1);
assert_eq!(log.append(&note("b")).unwrap(), 2);
assert_eq!(log.append(&note("c")).unwrap(), 3);
assert_eq!(log.append(note("a")).unwrap(), 1);
assert_eq!(log.append(note("b")).unwrap(), 2);
assert_eq!(log.append(note("c")).unwrap(), 3);
assert_eq!(log.last_seq(), 3);
let path = store.events_path_for(id);
@@ -211,9 +218,7 @@ fn append_assigns_line_position_as_seq_and_reads_back() {
let all = store.read_events(id, EventCursor::Since(0), 100).unwrap();
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 2, 3]);
assert_eq!(
all.iter()
.map(|e| text_of(&e.notification))
.collect::<Vec<_>>(),
all.iter().map(event_text).collect::<Vec<_>>(),
vec!["a", "b", "c"]
);
}
@@ -225,7 +230,7 @@ fn since_reads_forward_and_before_reads_backward_with_limit() {
let id = Uuid::new_v4();
let mut log = store.open_events(id).unwrap();
for i in 1..=6 {
log.append(&note(&format!("n{i}"))).unwrap();
log.append(note(&format!("n{i}"))).unwrap();
}
// `since` is exclusive: what a client that already holds seq 2 needs.
@@ -278,7 +283,7 @@ fn malformed_lines_are_skipped_but_keep_their_seq() {
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut log = store.open_events(id).unwrap();
log.append(&note("first")).unwrap();
log.append(note("first")).unwrap();
drop(log);
// Garbage in the middle and a line that is JSON but not a notification.
@@ -289,11 +294,11 @@ fn malformed_lines_are_skipped_but_keep_their_seq() {
std::fs::write(&path, raw).unwrap();
let mut log = store.open_events(id).unwrap();
assert_eq!(log.last_seq(), 3);
assert_eq!(log.append(&note("fourth")).unwrap(), 4);
assert_eq!(log.append(note("fourth")).unwrap(), 4);
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 4]);
assert_eq!(text_of(&all[1].notification), "fourth");
assert_eq!(event_text(&all[1]), "fourth");
}
#[test]
@@ -303,7 +308,7 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut log = store.open_events(id).unwrap();
log.append(&note("whole")).unwrap();
log.append(note("whole")).unwrap();
drop(log);
let path = store.events_path_for(id);
@@ -319,10 +324,10 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
// glue onto it; the partial keeps a seq of its own.
let mut log = store.open_events(id).unwrap();
assert_eq!(log.last_seq(), 2);
assert_eq!(log.append(&note("after")).unwrap(), 3);
assert_eq!(log.append(note("after")).unwrap(), 3);
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 3]);
assert_eq!(text_of(&all[1].notification), "after");
assert_eq!(event_text(&all[1]), "after");
let raw = std::fs::read_to_string(&path).unwrap();
assert_eq!(raw.lines().count(), 3, "{raw}");
}
@@ -334,8 +339,10 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() {
let id = Uuid::new_v4();
store.write(&sample(id)).unwrap();
let mut log = store.open_events(id).unwrap();
log.append(&note("x")).unwrap();
log.append(note("x")).unwrap();
drop(log);
let legacy_approvals = dir.path().join(format!("{id}.approvals.jsonl"));
std::fs::write(&legacy_approvals, b"obsolete").unwrap();
assert!(store.events_path_for(id).exists());
let loaded = store.load_all().unwrap();
@@ -344,6 +351,47 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() {
store.delete(id).unwrap();
assert!(!store.path_for(id).exists());
assert!(!store.events_path_for(id).exists());
assert!(!legacy_approvals.exists());
// Idempotent with the log already gone.
store.delete(id).unwrap();
}
#[test]
fn approval_rows_share_event_cursors_with_legacy_acp_lines() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let legacy = serde_json::to_vec(&note("before")).unwrap();
std::fs::create_dir_all(dir.path()).unwrap();
std::fs::write(store.events_path_for(id), [legacy, b"\n".to_vec()].concat()).unwrap();
let mut log = store.open_events(id).unwrap();
assert_eq!(
log.append(PersistedEvent::Approval {
row: TranscriptRow {
id: "approval:secret".into(),
kind: TranscriptRowKind::Approval,
data: serde_json::json!({"tool": "bash"}),
},
})
.unwrap(),
2
);
assert_eq!(log.append(note("after")).unwrap(), 3);
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
assert_eq!(
all.iter().map(|event| event.seq).collect::<Vec<_>>(),
[1, 2, 3]
);
assert!(matches!(all[1].event, PersistedEvent::Approval { .. }));
assert_eq!(event_text(&all[0]), "before");
assert_eq!(event_text(&all[2]), "after");
let since = store.read_events(id, EventCursor::Since(1), 1).unwrap();
assert_eq!(since[0].seq, 2);
let before = store
.read_events(id, EventCursor::Before(Some(3)), 1)
.unwrap();
assert_eq!(before[0].seq, 2);
}
+9
View File
@@ -872,6 +872,15 @@ pub struct StructuredPromptRequest {
pub prompt: String,
}
/// Operator decision for a pending ACP permission request.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StructuredApprovalDecision {
AllowOnce,
AllowForSession,
Deny,
}
// ---------------------------------------------------------------------------
// Client ↔ Daemon control messages (text frames on the attach WS)
// ---------------------------------------------------------------------------
+6
View File
@@ -91,6 +91,11 @@ pub fn structured_cancel(id: Uuid) -> String {
format!("{SESSIONS}/{id}/structured/cancel")
}
/// `POST /api/sessions/:id/structured/approvals/:nonce` — answer ACP permission.
pub fn structured_approval(id: Uuid, nonce: &str) -> String {
format!("{SESSIONS}/{id}/structured/approvals/{nonce}")
}
/// Axum-style patterns for the two parameterised routes, so the server
/// registers the same paths [`session`] and [`attach`] build.
pub mod patterns {
@@ -104,6 +109,7 @@ pub mod patterns {
pub const STRUCTURED_REPLAY: &str = "/api/sessions/:id/structured/replay";
pub const STRUCTURED_PROMPT: &str = "/api/sessions/:id/structured/prompt";
pub const STRUCTURED_CANCEL: &str = "/api/sessions/:id/structured/cancel";
pub const STRUCTURED_APPROVAL: &str = "/api/sessions/:id/structured/approvals/:nonce";
}
/// A daemon base URL that could not be turned into a WebSocket base.
+15 -5
View File
@@ -18,10 +18,14 @@ wizard, and cleans up — only the unified `maestro` binary ends up persisted in
**Updating is the same command.** Re-running `./install.sh` detects an existing
systemd unit, refreshes it, and runs `systemctl --user restart maestro` instead
of `start`. No `--update` flag needed. **The restart keeps your sessions:** the unit sets
of `start`. No `--update` flag needed. **The restart keeps your terminal sessions:** the unit sets
`KillMode=process`, so systemd signals only the daemon — the tmux server (and
every session) survives, and the new daemon re-adopts them on start. A unit
written before this default lacks the line; re-run `./install.sh` to refresh it.
every tmux-backed session) survives, and the new daemon re-adopts them on
start. Structured sessions (an agent run as a daemon child over ACP) are not
tmux panes, so shutdown ends their active turn; the transcript survives and
the next daemon attempts one resume (`docs/RESTART.md#structured-sessions`). A
unit written before this default lacks the line; re-run `./install.sh` to
refresh it.
Inside the script, `cargo install --git https://git.naps.pt/naps62/maestro …`
uses your already-configured git credentials for this host (HTTPS helper or
@@ -94,7 +98,10 @@ running a local daemon, for a host that only talks to another machine's.
The daemon unit carries the same `KillMode=process` as the wizard-written one,
so `systemctl --user restart maestro-daemon` leaves the tmux server and every
session alive.
terminal session alive. Structured sessions end their active turn on
shutdown and resume once on the next daemon: ACP state comes back via
`session/load`, while the event log only replays Maestro's own transcript
projection (`docs/RESTART.md#structured-sessions`).
## Thin builds for single-role hosts
@@ -211,7 +218,10 @@ what to run. The adapter is a child of the daemon: it ends with a `DELETE
itself. A daemon restart ends an in-flight structured turn, but preserves its
worktree, session record, and event-log transcript. The replacement daemon
starts a new adapter, resumes it with `session/load`, and replays the
transcript. Terminal sessions keep surviving restarts through tmux.
transcript. Terminal sessions keep surviving restarts through tmux. See
`docs/STRUCTURED-VIEW.md` for the full contract — row kinds, projections, the
handshake, and the approval and recovery rules — and `docs/RESTART.md` for
what a client observes across the restart.
## The browser UI
+39 -4
View File
@@ -2,10 +2,16 @@
Maestro is upgraded in place: the operator reinstalls the binary and systemd
restarts the daemon under every browser tab and TUI that is attached to it.
tmux keeps the sessions alive across that (see `docs/INSTALL.md`), but clients
lose their sockets. This document is the contract for what a client does about
it — what the daemon publishes, how a client tells a restart from a blip, and
what the user sees while it happens.
tmux keeps **terminal** sessions alive across that (see `docs/INSTALL.md`),
but clients lose their sockets either way. This document is the contract for
what a client does about it — what the daemon publishes, how a client tells a
restart from a blip, and what the user sees while it happens.
A **structured** session (`docs/STRUCTURED-VIEW.md`) has no tmux pane behind
it to survive on: its agent is a child of the daemon process itself, so a
restart takes it down with the daemon. See
[Structured sessions](#structured-sessions) below for exactly what that
costs and what it doesn't.
## What the daemon publishes
@@ -68,6 +74,35 @@ resolves, so a blocked client is never a trapped one. It clears itself the
moment a probe succeeds — there is no "retry" button, because the client is
already retrying.
## Structured sessions
Everything above describes a **client's** view of a restart, and it holds for
a structured session exactly as written — the same three verdicts, the same
status indicator, the same overlay. What differs is what survives on the
**daemon** side, because a structured session's agent is a daemon child, not
a tmux pane:
- A restart **ends the active turn**. There is no adapter process left to
finish it, and no tmux pane keeping it alive underneath.
- The **transcript survives**, in the per-session event log
(`docs/STRUCTURED-VIEW.md#persistence-recovery-and-dead-sessions`) — a
graceful shutdown marks the session for one resume attempt. Recovery is
two separate operations, not one: the next daemon calls `session/load` so
the adapter restores its own ACP session, but whatever history the
adapter emits during that call is discarded, not replayed into anything —
the daemon already has that history on disk. Separately, the daemon
rebuilds its own projection by folding the persisted event log back
through it. An ungraceful death (a crash, `kill -9`) leaves no resume
marker, and the session stays dead permanently.
- **Pending approvals and their nonces do not survive.** Both live only in
the old daemon's memory; any ACP permission request still mid-flight when
the daemon exits is answered `Cancelled` on the way down, and a resumed
session starts with no pending approval at all.
`docs/STRUCTURED-VIEW.md` is the full contract for all of this — the event
log's shape, the recovery marker, and exactly which kind of session death is
resumable.
## Timing
- **Grace window** — 1200 ms between the socket dropping and the overlay
+359
View File
@@ -0,0 +1,359 @@
# STRUCTURED-VIEW.md — the structured session contract
The contract for a structured session (`view: Structured`, #133): an agent run
as a daemon child over the Agent Client Protocol (ACP) instead of in a tmux
pane. Alongside `docs/WORKTREES.md`, `docs/SIDEBAR.md` and `docs/RESTART.md`,
read this before changing anything that renders a structured session — both
the web and TUI renderers are written against it.
It covers the eight transcript row kinds, what the daemon decides versus what
a client decides, the two projections and their delta/cursor semantics, the
ACP handshake and the capabilities it declines, the adapter's install
command, and the approval nonce rules. **Non-goal: the detached runner.** It
does not exist yet, and nothing here describes it.
## What a structured session is
A terminal session's agent runs inside a tmux pane; a structured session's
agent runs as a subprocess of the daemon itself, speaking ACP over stdio. The
daemon is the ACP **client**; the adapter process, `claude-code-acp`, is the
ACP **agent** — it wraps Claude Code's own SDK and translates the model's
tool calls into ACP `session/update` notifications instead of drawing them to
a pane. `crates/maestro-acp` owns the daemon's side of that connection;
`crates/maestro-daemon/src/projection.rs` folds what it receives into the two
projections below.
The child is a real child of the daemon process: same process group, and on
Linux marked with `PR_SET_PDEATHSIG` so it dies with the daemon even if the
daemon exits without cleaning up; other platforms have no such signal and
depend on an orderly shutdown/drop to terminate it. The ordinary path on
every platform is `AcpSession::terminate` (`session/cancel`, a short grace,
`SIGTERM`, then `SIGKILL`), called from both `DELETE /api/sessions/:id` and
daemon shutdown.
`docs/WORKTREES.md`'s kill-guard rule — an unforced kill refused while the
session's worktree holds unsaved work — applies to a structured session
exactly as it does to a terminal one; only what gets terminated differs (an
ACP handle instead of a tmux pane).
## The adapter
`ADAPTER_BIN` (`claude-code-acp`) is resolved by walking `PATH` — the
session's own `PATH` env var if the spawn request set one, the daemon's
otherwise — and taking the first executable file named `claude-code-acp` on
it. Install it once per host:
```sh
npm install -g --prefix ~/.local @zed-industries/claude-code-acp
```
That is `maestro_acp::INSTALL_COMMAND`, and it is not just documentation: it
is baked into `AcpError::AdapterMissing`'s message. When the adapter cannot
be found, spawn still creates the session — **Dead**, with that install
command as its `activity_detail` — so the sidebar names the fix instead of
the session silently never appearing (see `docs/INSTALL.md` for the full
install writeup, including the systemd `PATH` caveat). The adapter also needs
a logged-in Claude Code (`claude /login`); a missing login surfaces the same
way, via the adapter's own stderr tail in the failure message.
The handshake is bounded by `HANDSHAKE_TIMEOUT` (60s) — long enough for a
slow machine to boot a node process and the Claude Code SDK behind it, never
long enough to hang a spawn indefinitely. Any handshake failure — missing
adapter, early exit, a protocol error, or the timeout — leaves no orphaned
process: `spawn` closes and reaps the child itself before returning the
error.
## The ACP handshake
`initialize` is where the daemon states what kind of client it is. It
declares, deliberately, almost nothing:
| Capability | Declared | Why |
| --- | --- | --- |
| `fs.readTextFile` | `false` | The adapter uses Claude Code's own Read tool instead of asking the daemon to read a file on its behalf — the same tool a terminal-view Claude Code session uses. |
| `fs.writeTextFile` | `false` | Same reasoning, for Write. |
| `terminal` | `false` | The adapter runs Claude Code's own Bash tool rather than asking the daemon to open a terminal for it. |
Declining all three keeps the daemon out of the file/terminal surface
entirely — the adapter already has direct access to the session's `cwd`, and
routing every read, write, and shell command through the daemon would only
add a hop with no new capability. Every other capability namespace ACP
defines (session extensions, auth, elicitation) is simply left at its
default; the daemon does not advertise support because `claude-code-acp` has
never asked for it, not because it was weighed and refused.
The one client method that *is* implemented for real is
`session/request_permission` (#139) — see [Approvals](#approvals) below.
Before #139 it was a stub that declined every request; that stub is gone now
that a live implementation exists.
`session/new` opens a fresh ACP session on every ordinary spawn. A respawn
after a daemon restart instead sets `SpawnConfig::load_session(id)` and runs
`session/load` with the ACP session id the original handshake returned
(`Session::acp_session_id` on the wire) — the adapter must advertise
`agent_capabilities.load_session`, or the handshake fails outright naming
that. Whatever history the adapter streams back while `session/load` is in
flight is dropped on arrival, deliberately: the daemon's own event log
already holds that history, and forwarding it too would duplicate every row
in the projections below.
## The eight transcript row kinds
`TranscriptRowKind` (`maestro-protocol`) is the alphabet a structured
transcript is written in. Every row has a stable `id` (so later deltas can
address it), a `kind`, and a `data: Value` that preserves the underlying ACP
payload — clients depend on ACP's *shape* without depending on ACP as a
crate.
| Kind | Carries | Row identity |
| --- | --- | --- |
| `User` | An echoed `user_message_chunk` — the prompt text as ACP saw it. | `user:<message_id>`, or `user:<seq>` when the chunk carries no id. |
| `AgentText` | An `agent_message_chunk` — the agent's visible reply, streamed. | `agent:<message_id>` / `agent:<seq>`. Later chunks for the same id are appended into `data.content.text` in place, not appended as new rows. |
| `AgentThought` | An `agent_thought_chunk` — the model's reasoning stream. | `thought:<message_id>` / `thought:<seq>`, coalesced the same way as `AgentText`. |
| `ToolCall` | A `tool_call` / `tool_call_update` pair whose content carries no diff. | `tool:<tool_call_id>` — one row for the call's whole life; updates patch it in place, they never create a second row. |
| `ToolDiff` | The same tool-call machinery, when the tool's content includes a `Diff` block. | Same `tool:<tool_call_id>` identity as `ToolCall`; the *kind* flips between the two as the tool's own content changes, not the row. |
| `Approval` | A daemon-originated permission request (`RequestPermissionRequest`), serialized without its nonce. | `approval:<uuid>` — minted fresh per request, unrelated to the ACP request's own id. |
| `Plan` | A `plan` update — the agent's current step list. | The fixed id `"plan"`. Each new plan **replaces** the row wholesale; entries are not diffed against the previous plan. |
| `Notice` | The raw JSON of any `session/update` variant the projection does not special-case. | `notice:<seq>`. |
`Notice` is the projection's forward-compatibility seam: ACP adds
notification variants over time, and an unrecognised one becomes a generic
row instead of being silently dropped or making the daemon reject the whole
stream. A client that does not know what to do with a `Notice` can render it
as an opaque, collapsed line — that is a reasonable floor, not a client bug.
Rendering is entirely a client concern. The projection's job stops at
folding ACP into these eight shapes; markdown formatting, diff rendering,
tool-card treatment, and how a `Notice` is presented are all renderer
choices, not daemon ones.
## Daemon decisions vs. client decisions
**The daemon decides:**
- Which ACP `session/update` becomes which row kind, and under what id — the
fold in `projection.rs` is the one place this logic lives, so the web and
TUI renderers can never disagree about it.
- Turn lifecycle: when a turn begins, and whether an ACP response, a cancel,
or a failure ends it (`TurnState.responding` / `.thinking` /
`.active_tool_calls` are derived, not renderer-computed).
- Approval nonce minting, secrecy, and the single-active-approval rule (see
below) — a client never sees an ACP permission request before the daemon
has already turned it into an `Approval` row and a nonce.
- What happens to an ACP update variant nobody has taught the daemon about
yet (falls back to `Notice` rather than erroring).
- Persistence and replay: what is written to the event log, in what order,
and how a restart resumes it.
**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`.
- 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`),
never the UI that solicits it.
- Scrolling, collapsing, and every other presentation-only behaviour.
## The two projections
`Projection` (daemon-only) is folded once from the durable ACP stream and
exposes two independent views, both defined in `maestro-protocol`:
- **`TranscriptSnapshot { cursor, rows }`** — the complete ordered list of
transcript rows built so far, plus the log position (`cursor`) it reflects.
It is intentionally always the *whole* transcript, not a windowed one: a
tool row patched much later must still find its original entry.
- **`ReducedState`** — one whole-session snapshot folded over every event
seen: `turn` (the per-turn indicators above), `pending_approval`, `plan`,
`usage`/`context`, `model`, and `mode`. Like `TranscriptSnapshot`, it is
always full-history, never a delta — a client that only wants "is a turn
active right now" never has to replay anything to answer that.
Changes to the transcript travel as `TranscriptDelta`, tagged `type`:
`Append { row }` (a new row), `Patch { row }` (an existing row, addressed by
`id`, replaced wholesale), and `Remove { id }`. `Remove` is defined on the
wire for forward compatibility; nothing in the current projection emits it —
every row, once appended, is only ever patched or left alone.
### Cursor and replay semantics
The event log's own line position is the cursor: 1-based, shared by ACP
notifications and daemon-generated rows alike (`sidecar.rs`'s
`PersistedEvent` envelope is what lets an `Approval` row share a cursor
sequence with raw ACP JSON without pretending to be ACP traffic). Two ways to
read it:
- **`WS /api/structured/:id`** — on connect, sends a `TranscriptSnapshot` and
a `ReducedState` covering the whole log as it stands, then streams
`ProjectionUpdate { transcript, state }` for everything that happens next.
There is no client-supplied resume cursor on this socket: every connection
gets the full snapshot, not a gap-filled continuation, so a client never
needs the replay endpoint just to obtain the current transcript. Use
replay instead for cursor-bounded pagination through the durable log —
history from before this connection, fetched a page at a time.
- **`GET /api/sessions/:id/structured/replay`** — cursor-bounded pagination
over the durable log, returning one `StructuredReplayEvent` (`cursor`,
`transcript`, `state`) per matching line. `?since=<n>` returns the first
page after `n` (`since=0` selects from the beginning, i.e. the first
page); a client pages forward from there using the cursor of the last
event returned. `?before=<n>` returns the page immediately before `n`,
still in ascending order — passing both is a `400`. `?limit` defaults to
100 and is clamped to 500. Each call re-folds
the entire log from scratch to compute the state at every returned cursor,
so a page reflects exactly the *durable* state at its cursor, not an
approximation of it — at the cost of redoing the whole fold per request.
That durable state is not always identical to what a live client saw at
that moment, though: turn lifecycle transitions are broadcast live without
a durable event backing them (see [Prompt and cancel](#prompt-and-cancel-bounded-liveness)),
and a restored `Approval` row carries no pending nonce, because the nonce
never reached disk (see [Approvals](#approvals)) — replay is exact and
reconstructible, not a perfect record of everything the live view showed.
A session that is not a live structured session (terminal, or a dead
adapter) answers `409 not_structured` from either route rather than a
coherent-looking but stale reply.
## Approvals
An ACP `session/request_permission` callback becomes an `Approval` row plus a
pending decision, entirely on the daemon side:
- **Generation.** The daemon mints the nonce itself — a fresh
`Uuid::new_v4()` string per request — rather than using anything ACP's own
request carries. ACP permission callbacks have no request id of their own
that would double as one.
- **Live-only placement.** The nonce lives in exactly one place: the
in-memory `pending_approvals` map keyed by nonce, and its mirror in
`ReducedState.pending_approval.nonce` on the live wire. It is never part of
the `Approval` row's `data` — the row is `RequestPermissionRequest`
serialized *without* the nonce.
- **Secrecy from the agent and from durable history.** The adapter itself
never learns the nonce — it is a daemon-to-client secret, never sent back
over the ACP connection. It is likewise never written to the event log:
restoring an `Approval` row from disk (`Projection::restore_approval`)
produces a historical row with no `pending_approval` in state, because the
nonce that would make it "pending" never made it to disk in the first
place. A restart therefore cannot resume a pending approval — see
[Persistence](#persistence-recovery-and-dead-sessions).
- **One live approval at a time.** A new permission request mints its own
nonce, then cancels whatever was already pending (`cancel_pending_approvals`)
before installing itself as the new pending approval — ACP serializes
requests through one connection, so this only ever fires if an adapter
nonetheless overlaps two. Either way, the prior pending request is
cancelled/cleared before the new one is installed, so the newest request
wins.
- **Single-use and duplicate resolution.**
`POST /api/sessions/:id/structured/approvals/:nonce` **always answers
`204`** for a known, live structured session — whether the nonce is the
one currently pending, one already consumed by an earlier answer, or one
that never existed. Only an unknown or non-structured *session* answers
otherwise (`404` / `409`). This is deliberate: two clients racing to answer the same
approval must not turn a harmless duplicate click into a client-visible
error. Whichever `POST` reaches the daemon first removes the nonce from
`pending_approvals` and is the one whose decision actually reaches the
adapter; every later `POST` for that nonce is a no-op `204`.
- **Cancellation vs. denial.** A turn cancel or a session shutdown resolves
any pending approval with ACP's `Cancelled` outcome, not a `Deny` decision
— the adapter should read those as "the operation was withdrawn," not "the
operator refused it."
- **Decision mapping.** `AllowOnce` and `AllowForSession` select the ACP
option of kind `AllowOnce` / `AllowAlways` from the request's own option
list. `Deny` prefers a one-shot refusal (`RejectOnce`) and falls back to a
permanent one (`RejectAlways`) only when that is the sole rejection option
offered; if neither is present, it answers `Cancelled`.
## Prompt and cancel: bounded liveness
`POST /api/sessions/:id/structured/prompt` and `.../cancel` are serialized
against each other through one per-session `operation` lock
(`AcpEventStream::operation`), because ACP permission callbacks carry no
per-request id — the daemon's own notion of "the active turn" is the only
thing a callback can be attributed to, so at most one turn may ever be
in flight, and a cancelled turn must fully settle before the next prompt is
allowed to claim the daemon's idea of "the active turn."
- A prompt claims the turn and dispatches it to ACP as one operation. While a
turn is active, a second prompt gets `409 turn_in_flight` rather than
queuing — one active turn per session, never a hidden queue of prompts.
- Cancel sends `session/cancel`, clears any pending approval as `Cancelled`,
and ends the turn's client-visible lifecycle immediately. If the cancelled
turn was still *dispatched* to ACP (waiting on its response), the daemon
cannot simply forget it: ACP has not confirmed the cancel yet, and a new
prompt claiming "the active turn" while the old one is still outstanding
would make the next permission callback ambiguous between the two.
- So a cancelled-but-still-dispatched turn must **settle within
`CANCEL_GRACE` (500ms)** — either ACP acknowledges the prompt (however it
resolves) or that grace period runs out. If it settles in time, the next
prompt proceeds normally. If it does not, the daemon does not guess: it
marks the whole structured session dead (`fail_cancelled_acp_turn`) rather
than risk a callback resolving against the wrong turn. This is the bounded
half of "bounded liveness" — a structured session's prompt/cancel dance
either resolves promptly or the session is declared unusable, never left
in an indefinite ambiguous state.
## Persistence, recovery, and dead sessions
A structured session's durable state is three files in the sidecar
directory: the sidecar itself (`<id>.json`), plus two next to it:
- **`<id>.events.jsonl`** — the unified event log. Each line is either a raw
ACP `SessionNotification` (kept in its original JSON so old logs stay
readable) or a `{"maestroEvent": "approval", "row": …}` envelope for a
daemon-generated row. Both share the same 1-based line-position cursor. A
line a hard kill left half-written is terminated before the next append, so
it keeps its position and is skipped as malformed rather than corrupting
what follows.
- **`<id>.recover`** — an empty marker file. Its only meaning is "the daemon
that wrote this shut down gracefully while this session was running, and
the next daemon may attempt to resume it." It is consumed (deleted) the
moment a recovery attempt reads it — one recovery attempt per graceful
shutdown, never a standing flag.
Three ways a structured session ends up `Dead`, each with different
persistence consequences:
1. **The adapter exits, or a notification append to the event log fails,
while the daemon keeps running.** (An approval-row append failure is
narrower: it cancels the pending approval and lets the session continue,
without marking it dead.) `mark_session_dead` applies the same
natural-exit worktree rule
as a terminal session's own pane exiting (`docs/WORKTREES.md`): a pristine
worktree is removed, one holding unsaved work is left on disk. No recovery
marker is written. This death is permanent — a later daemon restart's
`recover()` finds the sidecar, sees no marker, and leaves the session
`Dead` without attempting to resume it.
2. **A cancelled turn fails to settle within `CANCEL_GRACE`.** Same
consequence as (1): the session is marked dead through the same path, no
marker, no future resumption.
3. **Graceful daemon shutdown**, while the session was still `Running`.
`mark_session_dead_after_shutdown` marks the session dead **without**
applying the worktree exit rule — the tree is kept exactly as it is,
pristine or not, because there is nothing "natural" about this exit — and
writes the `.recover` marker. The sidecar on disk therefore says `Dead`,
but is eligible for resumption.
Recovery, on the next daemon's `recover()`: for every `Structured` sidecar
not already adopted, the marker is consumed. When one was present and the
sidecar carries a stored `acp_session_id`, the daemon spawns a fresh adapter
with `SpawnConfig::load_session(acp_session_id)`, replays the
entire durable event log into a new `Projection` (rebuilding the transcript
and reduced state exactly as they were, with any restored `Approval` row
landing as history — never as a re-armed pending approval, since its nonce
never reached disk), and only then reopens the log for further appends. What
the adapter streams back while its own `session/load` is in flight is
discarded rather than appended, so the replayed history is not duplicated
(see [The ACP handshake](#the-acp-handshake)). Success flips the session back
to `Running`/`Active`; failure (no stored `acp_session_id`, or the handshake
itself failing) leaves it `Dead` with the failure reason as
`activity_detail` — the same "tell the operator what to fix" pattern as a
missing adapter binary.
Pending approvals and their nonces never survive any of this. They live only
in the in-memory `pending_approvals` map and the live `ReducedState`, both
gone the instant the daemon that held them exits. Whatever ACP permission
request was mid-flight when the old daemon died was already answered
`Cancelled` on that daemon's own shutdown path, so a resumed session always
starts with no pending approval.
See `docs/RESTART.md` for what a *client* observes across a daemon restart,
now that structured sessions exist alongside terminal ones.