From 3c638b08de0cfba43dfa1a9f8d8e83635ae16c05 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 5 Sep 2026 02:05:06 +0100 Subject: [PATCH] fix: clarify structured prompt failures --- crates/maestro-acp/tests/fake_acp_agent.py | 9 ++- crates/maestro-daemon/src/local_api.rs | 11 +++- crates/maestro-daemon/src/session.rs | 24 +++++-- crates/maestro-daemon/tests/acp.rs | 38 +++++------ crates/maestro-daemon/tests/local_api.rs | 77 ++++++++++++++++++++-- 5 files changed, 124 insertions(+), 35 deletions(-) diff --git a/crates/maestro-acp/tests/fake_acp_agent.py b/crates/maestro-acp/tests/fake_acp_agent.py index b1a379e..f05aacf 100755 --- a/crates/maestro-acp/tests/fake_acp_agent.py +++ b/crates/maestro-acp/tests/fake_acp_agent.py @@ -30,7 +30,9 @@ Environment knobs, all optional: the way a real agent streams history back FAKE_ACP_LOAD_FAIL=1 answer `session/load` with a JSON-RPC error FAKE_ACP_DELAY_PROMPT_MS=n wait before answering `session/prompt` - FAKE_ACP_PROMPT_FAIL=1 answer `session/prompt` with a JSON-RPC error + 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 Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick ", counting from 1) and logs `emit tick `, so a test can make the agent @@ -67,6 +69,7 @@ signal.signal(signal.SIGTERM, on_sigterm) session_id = f"fake-{os.getpid()}" ticks = 0 +prompt_failures = 0 def chunk(text): @@ -141,7 +144,9 @@ for raw in sys.stdin: sys.exit(0) elif method == "session/prompt": time.sleep(int(os.environ.get("FAKE_ACP_DELAY_PROMPT_MS", "0")) / 1000) - if os.environ.get("FAKE_ACP_PROMPT_FAIL") == "1": + 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): + prompt_failures += 1 send({"jsonrpc": "2.0", "id": rid, "error": {"code": -32000, "message": "fake prompt failure"}}) else: diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index e7959c4..030519d 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -433,14 +433,21 @@ async fn structured_prompt( Json(body): Json, ) -> Response { let (acp, turn) = match s.manager.begin_acp_turn(id).await { - Ok(Some(turn)) => turn, - Ok(None) => { + 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", + ) + } Err(e) => { let info = e.to_error_info(); return error_response(session_error_status(&info.code), &info.code, &info.message); diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index a3a4340..246a125 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -139,6 +139,14 @@ pub enum SessionError { KillGuarded(maestro_protocol::InTheWay), } +/// The result of claiming the next prompt turn for a structured ACP session. +#[derive(Debug)] +pub enum BeginAcpTurn { + Started(Arc, u64), + InFlight, + Unavailable, +} + impl SessionError { pub fn to_error_info(&self) -> ErrorInfo { let code = match self { @@ -476,25 +484,27 @@ impl SessionManager { /// Begin an ACP turn only when the previous one has ended. The projection /// lock makes the rejection and lifecycle update one indivisible action. - pub async fn begin_acp_turn( - &self, - id: Uuid, - ) -> Result, u64)>, SessionError> { + pub async fn begin_acp_turn(&self, id: Uuid) -> Result { let (acp, stream) = { let sessions = self.inner.sessions.lock().await; let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?; + if record.session.view != SessionView::Structured + || record.session.status != SessionStatus::Running + { + return Ok(BeginAcpTurn::Unavailable); + } (record.acp.clone(), record.acp_events.clone()) }; let (Some(acp), Some(stream)) = (acp, stream) else { - return Ok(None); + return Ok(BeginAcpTurn::Unavailable); }; let mut stream = stream.lock().await; if stream.projection.turn_active() { - return Ok(None); + return Ok(BeginAcpTurn::InFlight); } let (turn, update) = stream.projection.begin_turn(); let _ = stream.projected_live.send(update); - Ok(Some((acp, turn))) + Ok(BeginAcpTurn::Started(acp, turn)) } /// Cancel an ACP turn and immediately clear its client-facing lifecycle. diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index 5340bb0..f707eb4 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -232,12 +232,14 @@ async fn projection_snapshot_and_live_stream_share_one_fold() { assert_eq!(snapshot.rows[0].kind, TranscriptRowKind::AgentText); assert_eq!(state.turn, Default::default()); - let (_, turn) = fx + let maestro_daemon::session::BeginAcpTurn::Started(_, turn) = fx .manager .begin_acp_turn(session.id) .await .expect("begin lifecycle") - .expect("structured turn"); + else { + panic!("structured turn should start"); + }; let begin = tokio::time::timeout(Duration::from_secs(5), live.recv()) .await .expect("begin lifecycle broadcast") @@ -289,31 +291,27 @@ async fn stale_turn_completion_cannot_end_the_next_turn() { .await .unwrap(); - let (_, first) = fx - .manager - .begin_acp_turn(session.id) - .await - .unwrap() - .unwrap(); + let maestro_daemon::session::BeginAcpTurn::Started(_, first) = + fx.manager.begin_acp_turn(session.id).await.unwrap() + else { + panic!("first turn should start"); + }; assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap()); - let (_, second) = fx - .manager - .begin_acp_turn(session.id) - .await - .unwrap() - .unwrap(); + let maestro_daemon::session::BeginAcpTurn::Started(_, second) = + fx.manager.begin_acp_turn(session.id).await.unwrap() + else { + panic!("second turn should start"); + }; assert!(second > first); assert!(!fx .manager .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, first) .await .unwrap()); - assert!(fx - .manager - .begin_acp_turn(session.id) - .await - .unwrap() - .is_none()); + assert!(matches!( + fx.manager.begin_acp_turn(session.id).await.unwrap(), + maestro_daemon::session::BeginAcpTurn::InFlight + )); assert!(fx .manager .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, second) diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 188911c..8aea8d3 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -10,7 +10,9 @@ use std::time::Duration; use common::DaemonFixture; use futures_util::{SinkExt, StreamExt}; use maestro_daemon::local_api; -use maestro_protocol::{ApiError, CreateSessionRequest, Session, SessionView, WorktreeSpec}; +use maestro_protocol::{ + ApiError, CreateSessionRequest, Session, SessionStatus, SessionView, WorktreeSpec, +}; use tempfile::TempDir; use tokio::net::TcpListener; use tokio::sync::oneshot; @@ -261,10 +263,15 @@ async fn structured_prompt_failure_clears_the_turn() { let h = spawn("naps-laptop").await; let fake = FakeAcp::new(); let mut env = fake.env(); - env.insert("FAKE_ACP_PROMPT_FAIL".into(), "1".into()); + env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into()); let id = structured_session_with_env(&h, env).await; let client = reqwest::Client::new(); let path = format!("/api/sessions/{id}/structured/prompt"); + 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(); let first = client .post(url(h.addr, &path)) .json(&serde_json::json!({"prompt":"fails"})) @@ -272,14 +279,76 @@ async fn structured_prompt_failure_clears_the_turn() { .await .unwrap(); assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT); - tokio::time::sleep(Duration::from_millis(50)).await; + let mut saw_active_turn = false; + tokio::time::timeout(Duration::from_secs(1), async { + while let Some(Ok(message)) = ws.next().await { + let event: serde_json::Value = + serde_json::from_str(&message.into_text().unwrap()).unwrap(); + let turn = &event["state"]["turn"]; + if turn["responding"] == true { + saw_active_turn = true; + } + if saw_active_turn && turn["responding"] == false && turn["thinking"] == false { + return; + } + } + panic!("structured websocket closed before the failed turn cleared"); + }) + .await + .expect("failed turn clears promptly"); let second = client .post(url(h.addr, &path)) .json(&serde_json::json!({"prompt":"available"})) .send() .await .unwrap(); - assert_ne!(second.status(), reqwest::StatusCode::CONFLICT); + assert_eq!(second.status(), reqwest::StatusCode::NO_CONTENT); + assert!(second.bytes().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn structured_prompt_rejects_unknown_and_unavailable_sessions() { + let h = spawn("naps-laptop").await; + let client = reqwest::Client::new(); + let prompt = |id| { + client + .post(url( + h.addr, + &format!("/api/sessions/{id}/structured/prompt"), + )) + .json(&serde_json::json!({"prompt": "hello"})) + }; + + let unknown = prompt(Uuid::new_v4()).send().await.unwrap(); + assert_eq!(unknown.status(), reqwest::StatusCode::NOT_FOUND); + let error: ApiError = unknown.json().await.unwrap(); + assert_eq!(error.error.code, "not_found"); + + let terminal = h + .manager + .create(CreateSessionRequest { + view: SessionView::Terminal, + cwd: "/tmp".into(), + command: "sh".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .unwrap(); + let non_structured = prompt(terminal.id).send().await.unwrap(); + assert_eq!(non_structured.status(), reqwest::StatusCode::CONFLICT); + let error: ApiError = non_structured.json().await.unwrap(); + assert_eq!(error.error.code, "not_structured"); + + let fake = FakeAcp::new(); + let dead = structured_session(&h, &fake).await; + h.manager.shutdown().await; + assert_eq!(h.manager.status(dead).await, Some(SessionStatus::Dead)); + let unavailable = prompt(dead).send().await.unwrap(); + assert_eq!(unavailable.status(), reqwest::StatusCode::CONFLICT); + let error: ApiError = unavailable.json().await.unwrap(); + assert_eq!(error.error.code, "not_structured"); } #[tokio::test]