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

fix: clarify structured prompt failures

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 02:05:06 +01:00
parent be23a2b539
commit 3c638b08de
5 changed files with 124 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-acp/tests/fake_acp_agent.py
+7 -2
View File
@@ -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 <n>",
counting from 1) and logs `emit tick <n>`, 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:
crates/maestro-daemon/src/local_api.rs
+9 -2
View File
@@ -433,14 +433,21 @@ async fn structured_prompt(
Json(body): Json<StructuredPromptRequest>,
) -> 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);
crates/maestro-daemon/src/session.rs
+17 -7
View File
@@ -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<AcpSession>, 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<Option<(Arc<AcpSession>, u64)>, SessionError> {
pub async fn begin_acp_turn(&self, id: Uuid) -> Result<BeginAcpTurn, SessionError> {
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.
crates/maestro-daemon/tests/acp.rs
+18 -20
View File
@@ -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)
crates/maestro-daemon/tests/local_api.rs
+73 -4
View File
@@ -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]