Compare commits

..
Author SHA1 Message Date
Miguel Palhas 74a423567f docs(maestro): fix disabled-unstable-variant claim in STRUCTURED-VIEW.md
A SessionUpdate variant gated behind an unstable ACP feature this build
doesn't enable is compiled out of the enum, not decoded into a Notice.
It fails deserialization the same as a genuinely unknown future tag.
2026-09-05 05:35:19 +01:00
Miguel Palhas b225fbdd69 docs(maestro): narrow the Notice forward-compat claim in STRUCTURED-VIEW.md
Notice only covers SessionUpdate variants the agent-client-protocol
dependency already decodes but the projection doesn't special-case.
SessionUpdate is an internally-tagged enum with no unknown-variant
fallback, so a genuinely new wire variant fails deserialization before
the projection ever sees it, rather than becoming a Notice. A new
variant needs the dependency upgraded first.

Adds a projection test asserting that an unrecognised sessionUpdate
tag is a deserialization error.

Refs #186
2026-09-05 05:28:11 +01:00
Miguel Palhas 244e8d2927 Merge structured-view/185-lock-order into blitz/structured-view 2026-09-05 05:20:20 +01:00
Miguel Palhas 4a1a10c49a fix(maestro): avoid nested session locks 2026-09-05 05:15:58 +01:00
Miguel Palhas d607740fea Merge structured-view/184-approval-persist into blitz/structured-view 2026-09-05 05:05:15 +01:00
3 changed files with 226 additions and 13 deletions
+10
View File
@@ -438,6 +438,16 @@ mod tests {
.unwrap()
}
#[test]
fn a_session_update_tag_the_acp_dependency_does_not_know_fails_to_deserialize() {
// `SessionUpdate` has no unknown-variant fallback, so this never reaches `Projection::apply`.
let result: Result<SessionNotification, _> = serde_json::from_value(serde_json::json!({
"sessionId": "acp-1",
"update": {"sessionUpdate": "some_future_update_variant"},
}));
assert!(result.is_err());
}
#[test]
fn folds_content_tool_diff_plan_and_notice_rows() {
let mut projection = Projection::default();
+193 -5
View File
@@ -2003,6 +2003,41 @@ fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> Pr
}
async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
let events = {
let sessions = inner.sessions.lock().await;
let Some(record) = sessions.get(&id) else {
return;
};
if record.session.view != SessionView::Structured {
return;
}
record.acp_events.clone()
};
match events {
Some(events) => sync_structured_activity_from_stream(inner, id, events).await,
None => apply_structured_activity(inner, id, None, ActivityState::Error).await,
}
}
async fn sync_structured_activity_from_stream(
inner: &Arc<Inner>,
id: Uuid,
events: Arc<Mutex<AcpEventStream>>,
) {
let event_stream = events.lock().await;
let raw = event_stream.projection.activity();
// Stream state must stay fixed until its catalog transition is applied;
// every other path releases `sessions` before awaiting this stream lock.
apply_structured_activity(inner, id, Some(&events), raw).await;
}
async fn apply_structured_activity(
inner: &Arc<Inner>,
id: Uuid,
expected_events: Option<&Arc<Mutex<AcpEventStream>>>,
raw: ActivityState,
) {
let updated = {
let mut sessions = inner.sessions.lock().await;
let Some(record) = sessions.get_mut(&id) else {
@@ -2011,12 +2046,19 @@ async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
if record.session.view != SessionView::Structured {
return;
}
let current_events = record.acp_events.as_ref();
let stream_matches = match (expected_events, current_events) {
(Some(expected), Some(current)) => Arc::ptr_eq(expected, current),
(None, None) => true,
_ => false,
};
if !stream_matches {
return;
}
let raw = if record.session.status == SessionStatus::Dead {
ActivityState::Error
} else if let Some(events) = &record.acp_events {
events.lock().await.projection.activity()
} else {
ActivityState::Error
raw
};
let activity = crate::activity::apply_focus_suppression(
raw,
@@ -2702,20 +2744,28 @@ async fn cleanup_record(
#[cfg(test)]
mod tests {
use std::fs::File;
use std::sync::Arc;
use std::task::Poll;
use std::time::Duration;
use maestro_acp::schema::{
PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest,
ToolCallUpdate, ToolCallUpdateFields,
};
use maestro_protocol::{ActivityState, CreateSessionRequest, SessionView};
use super::{
cancel_pending_approvals, permission_outcome, record_permission_request, AcpDispatchState,
AcpEventStream, ApprovalDecision, RecordOutcome,
cancel_pending_approvals, permission_outcome, record_permission_request,
run_structured_activity_pass, sync_structured_activity_from_stream, AcpDispatchState,
AcpEventStream, ApprovalDecision, RecordOutcome, SessionManager,
};
use crate::activity::FOCUS_CLEAR_HOLD;
use crate::panes::FakePanes;
use crate::projection::Projection;
use crate::sidecar::{
EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore,
};
use crate::worktree::WorktreeManager;
fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
PermissionOption::new(id.to_owned(), id, kind)
@@ -2743,6 +2793,144 @@ mod tests {
}
}
fn terminal_request(cwd: &std::path::Path) -> CreateSessionRequest {
CreateSessionRequest {
view: SessionView::Terminal,
cwd: cwd.to_path_buf(),
command: "sh".into(),
env: std::collections::HashMap::new(),
metadata: std::collections::HashMap::new(),
worktree: None,
}
}
async fn manager_with_structured_stream() -> (
tempfile::TempDir,
SessionManager,
uuid::Uuid,
Arc<tokio::sync::Mutex<AcpEventStream>>,
uuid::Uuid,
) {
let state = tempfile::tempdir().unwrap();
let manager = SessionManager::new(
Arc::new(FakePanes::new()),
SidecarStore::new(state.path().join("sessions")),
WorktreeManager::new(None),
state.path().join("fifos"),
);
let structured = manager
.create(terminal_request(state.path()))
.await
.expect("create structured fixture");
let terminal = manager
.create(terminal_request(state.path()))
.await
.expect("create unrelated terminal");
let events = Arc::new(tokio::sync::Mutex::new(stream(
manager
.inner
.sidecar
.open_events(structured.id)
.expect("open event log"),
)));
{
let mut event_stream = events.lock().await;
event_stream.projection.begin_turn();
event_stream
.projection
.request_approval("nonce".into(), serde_json::json!({}));
}
{
let mut sessions = manager.inner.sessions.lock().await;
let record = sessions.get_mut(&structured.id).unwrap();
record.session.view = SessionView::Structured;
record.session.activity = ActivityState::AwaitingInput;
record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD);
record.acp_events = Some(events.clone());
}
(state, manager, structured.id, events, terminal.id)
}
#[tokio::test]
async fn blocked_structured_activity_read_does_not_block_the_catalog() {
let (state, manager, structured_id, events, terminal_id) =
manager_with_structured_stream().await;
let held_stream = events.lock().await;
let activity_pass = run_structured_activity_pass(&manager.inner);
tokio::pin!(activity_pass);
assert!(matches!(
futures_util::poll!(activity_pass.as_mut()),
Poll::Pending
));
tokio::time::timeout(Duration::from_secs(1), manager.snapshot())
.await
.expect("catalog snapshot blocked behind another session's event stream");
tokio::time::timeout(
Duration::from_secs(1),
manager.send_keys(terminal_id, b"responsive"),
)
.await
.expect("terminal input blocked behind another session's event stream")
.expect("terminal input");
let created = tokio::time::timeout(
Duration::from_secs(1),
manager.create(terminal_request(state.path())),
)
.await
.expect("session create blocked behind another session's event stream")
.expect("create unrelated session");
tokio::time::timeout(Duration::from_secs(1), manager.kill(created.id, false))
.await
.expect("session kill blocked behind another session's event stream")
.expect("kill unrelated session");
drop(held_stream);
tokio::time::timeout(Duration::from_secs(1), activity_pass)
.await
.expect("activity pass did not resume");
let structured = manager
.snapshot()
.await
.into_iter()
.find(|session| session.id == structured_id)
.unwrap();
assert_eq!(structured.activity, ActivityState::Idle);
}
#[tokio::test]
async fn delayed_activity_from_a_replaced_stream_is_discarded() {
let (_state, manager, structured_id, old_events, _terminal_id) =
manager_with_structured_stream().await;
let replacement = Arc::new(tokio::sync::Mutex::new(stream(
manager
.inner
.sidecar
.open_events(uuid::Uuid::new_v4())
.expect("open replacement event log"),
)));
let mut sessions = manager.inner.sessions.lock().await;
let sync =
sync_structured_activity_from_stream(&manager.inner, structured_id, old_events.clone());
tokio::pin!(sync);
assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending));
let record = sessions.get_mut(&structured_id).unwrap();
record.acp_events = Some(replacement.clone());
record.session.activity = ActivityState::Error;
drop(sessions);
tokio::time::timeout(Duration::from_secs(1), sync)
.await
.expect("stale activity sync did not finish");
let sessions = manager.inner.sessions.lock().await;
let record = sessions.get(&structured_id).unwrap();
assert_eq!(record.session.activity, ActivityState::Error);
assert!(Arc::ptr_eq(
record.acp_events.as_ref().unwrap(),
&replacement
));
}
/// A handle that can never append *or* roll back: opened read-only, so
/// both `write_all` and the rollback `truncate` fail with the OS's own
/// error (`ftruncate` requires a writable fd too) instead of a
+23 -8
View File
@@ -113,13 +113,25 @@ crate.
| `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` | The raw JSON of any `session/update` variant the ACP dependency already decodes but 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.
`Notice` covers known-but-unspecialized `SessionUpdate` variants: ones the
`agent-client-protocol` dependency's schema already deserializes (it
currently folds `available_commands_update` and `session_info_update`
there) but the projection's `match` doesn't fold into one of the other
seven kinds. It is not a wire-level forward-compatibility seam:
`SessionUpdate` is an internally-tagged enum with no unknown-variant
fallback, so a `session/update` whose `sessionUpdate` tag this build's
dependency has never heard of — including a variant gated behind an
unstable ACP feature this build doesn't enable, which is compiled out of
the enum entirely — fails to deserialize before the projection ever runs,
rather than becoming a `Notice`. A genuinely new ACP variant needs the
`agent-client-protocol` dependency upgraded (and, typically, the projection's
`match` taught about it) before it can appear on this transcript at all — see
`projection.rs`'s test asserting that an unrecognised `sessionUpdate` tag is a
deserialization error, not a `Notice`. A client that does not know what to do
with a `Notice` can still render it as an opaque, collapsed line — that
remains 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,
@@ -139,8 +151,11 @@ choices, not daemon ones.
- 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).
- What happens to an ACP update variant the dependency decodes but nobody has
taught the projection to special-case (falls back to `Notice` rather than
erroring). A variant the dependency itself can't decode is a deserialization
error, not a `Notice` — see [The eight transcript row
kinds](#the-eight-transcript-row-kinds).
- Persistence and replay: what is written to the event log, in what order,
and how a restart resumes it.