Miguel PalhasandClaude Opus 4.6
ebe25a1a55
Prompts sent while a turn is active are queued on the daemon instead of in the browser's React state. The queue is persisted in the session sidecar and drained as the next turn when the current one completes. Fixes the bug where queued messages were delivered to the wrong session when switching focus in the UI: the client-side queue captured the new sessionId in its drain closure. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1607 lines
50 KiB
Rust
1607 lines
50 KiB
Rust
//! Integration tests for the daemon's HTTP + WS API (NAP-243).
|
|
|
|
mod common;
|
|
|
|
use std::collections::HashMap;
|
|
use std::net::{Ipv4Addr, SocketAddr};
|
|
use std::path::Path;
|
|
use std::time::Duration;
|
|
|
|
use common::DaemonFixture;
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use maestro_daemon::local_api;
|
|
use maestro_protocol::{
|
|
ActivityState, ApiError, CreateSessionRequest, Session, SessionStatus, SessionView,
|
|
WorktreeSpec,
|
|
};
|
|
use tempfile::TempDir;
|
|
use tokio::net::TcpListener;
|
|
use tokio::sync::oneshot;
|
|
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
|
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
|
|
use uuid::Uuid;
|
|
|
|
type WsClient = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
|
|
|
|
struct Harness {
|
|
addr: SocketAddr,
|
|
manager: maestro_daemon::SessionManager,
|
|
_shutdown: oneshot::Sender<()>,
|
|
_fx: DaemonFixture,
|
|
}
|
|
|
|
async fn spawn(hostname: &str) -> Harness {
|
|
spawn_with(
|
|
hostname,
|
|
std::sync::Arc::new(maestro_daemon::projects::ProjectScanner::disabled()),
|
|
)
|
|
.await
|
|
}
|
|
|
|
async fn spawn_with(
|
|
hostname: &str,
|
|
projects: std::sync::Arc<maestro_daemon::projects::ProjectScanner>,
|
|
) -> Harness {
|
|
spawn_with_leader(hostname, projects, maestro_config::DEFAULT_LEADER).await
|
|
}
|
|
|
|
async fn spawn_with_leader(
|
|
hostname: &str,
|
|
projects: std::sync::Arc<maestro_daemon::projects::ProjectScanner>,
|
|
leader: &str,
|
|
) -> Harness {
|
|
let fx = DaemonFixture::new();
|
|
let manager = fx.manager.clone();
|
|
// Built-in presets (claude/opencode/zsh + custom) so `/api/info` advertises
|
|
// a known set (#26).
|
|
let presets = maestro_config::SpawnSection::default().resolve().unwrap();
|
|
let router = local_api::router(local_api::state_for(
|
|
manager.clone(),
|
|
hostname,
|
|
"/home/test",
|
|
presets,
|
|
projects,
|
|
leader,
|
|
));
|
|
let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
|
|
.await
|
|
.unwrap();
|
|
let addr = listener.local_addr().unwrap();
|
|
let (tx, rx) = oneshot::channel::<()>();
|
|
tokio::spawn(async move {
|
|
let _ = axum::serve(listener, router)
|
|
.with_graceful_shutdown(async move {
|
|
let _ = rx.await;
|
|
})
|
|
.await;
|
|
});
|
|
Harness {
|
|
addr,
|
|
manager,
|
|
_shutdown: tx,
|
|
_fx: fx,
|
|
}
|
|
}
|
|
|
|
fn url(addr: SocketAddr, path: &str) -> String {
|
|
format!("http://{addr}{path}")
|
|
}
|
|
|
|
fn ws_url(addr: SocketAddr, path: &str) -> String {
|
|
format!("ws://{addr}{path}")
|
|
}
|
|
|
|
struct FakeAcp {
|
|
dir: TempDir,
|
|
}
|
|
|
|
impl FakeAcp {
|
|
fn new() -> Self {
|
|
use std::os::unix::fs::PermissionsExt;
|
|
|
|
let dir = TempDir::new().unwrap();
|
|
let bin = dir.path().join("bin");
|
|
std::fs::create_dir_all(&bin).unwrap();
|
|
let adapter = bin.join("claude-code-acp");
|
|
std::fs::copy(
|
|
Path::new(env!("CARGO_MANIFEST_DIR")).join("../maestro-acp/tests/fake_acp_agent.py"),
|
|
&adapter,
|
|
)
|
|
.unwrap();
|
|
std::fs::set_permissions(&adapter, std::fs::Permissions::from_mode(0o755)).unwrap();
|
|
Self { dir }
|
|
}
|
|
|
|
fn env(&self) -> HashMap<String, String> {
|
|
HashMap::from([
|
|
(
|
|
"PATH".into(),
|
|
format!(
|
|
"{}:{}",
|
|
self.dir.path().join("bin").display(),
|
|
std::env::var("PATH").unwrap()
|
|
),
|
|
),
|
|
(
|
|
"FAKE_ACP_LOG".into(),
|
|
self.dir.path().join("agent.log").display().to_string(),
|
|
),
|
|
])
|
|
}
|
|
|
|
fn pid(&self) -> u32 {
|
|
std::fs::read_to_string(self.dir.path().join("agent.log"))
|
|
.unwrap_or_default()
|
|
.lines()
|
|
.filter_map(|line| {
|
|
line.strip_prefix("start pid=")?
|
|
.split(' ')
|
|
.next()?
|
|
.parse()
|
|
.ok()
|
|
})
|
|
.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 {
|
|
structured_session_with_env(h, fake.env()).await
|
|
}
|
|
|
|
async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>) -> Uuid {
|
|
let id = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: SessionView::Structured,
|
|
cwd: "/tmp".into(),
|
|
command: "claude".into(),
|
|
env,
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap()
|
|
.id;
|
|
wait_for_session_activity(h, id, ActivityState::Idle).await;
|
|
id
|
|
}
|
|
|
|
async fn wait_for_session_activity(h: &Harness, id: Uuid, expected: ActivityState) {
|
|
tokio::time::timeout(Duration::from_secs(2), async {
|
|
loop {
|
|
let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
if sessions
|
|
.iter()
|
|
.find(|session| session.id == id)
|
|
.is_some_and(|session| session.activity == expected)
|
|
{
|
|
return;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap_or_else(|_| panic!("session {id} did not reach {expected:?}"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn structured_ws_sends_snapshot_then_live_projection() {
|
|
let h = spawn("naps-laptop").await;
|
|
let fake = FakeAcp::new();
|
|
let id = structured_session(&h, &fake).await;
|
|
let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
|
|
.await
|
|
.unwrap();
|
|
|
|
let first = ws.next().await.unwrap().unwrap().into_text().unwrap();
|
|
let second = ws.next().await.unwrap().unwrap().into_text().unwrap();
|
|
assert_eq!(
|
|
serde_json::from_str::<serde_json::Value>(&first).unwrap()["type"],
|
|
"transcript_snapshot"
|
|
);
|
|
assert_eq!(
|
|
serde_json::from_str::<serde_json::Value>(&second).unwrap()["type"],
|
|
"reduced_state"
|
|
);
|
|
|
|
let response = reqwest::Client::new()
|
|
.post(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/prompt"),
|
|
))
|
|
.json(&serde_json::json!({"prompt": "hello"}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
|
|
let live = tokio::time::timeout(Duration::from_secs(1), ws.next())
|
|
.await
|
|
.expect("projection update arrives on the subscribed stream")
|
|
.unwrap()
|
|
.unwrap()
|
|
.into_text()
|
|
.unwrap();
|
|
assert_eq!(
|
|
serde_json::from_str::<serde_json::Value>(&live).unwrap()["type"],
|
|
"projection_update"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn structured_attachments_preserve_bytes_and_reject_paths() {
|
|
let h = spawn("attachments").await;
|
|
let fake = FakeAcp::new();
|
|
let id = structured_session(&h, &fake).await;
|
|
let client = reqwest::Client::new();
|
|
let endpoint = url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/attachments"),
|
|
);
|
|
let bytes = vec![0, 255, 10, 13];
|
|
let response = client
|
|
.post(&endpoint)
|
|
.query(&[("name", "sample.bin")])
|
|
.body(bytes.clone())
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), 200);
|
|
let body: serde_json::Value = response.json().await.unwrap();
|
|
let path = body["path"].as_str().unwrap();
|
|
assert!(Path::new(path).is_absolute());
|
|
assert_eq!(std::fs::read(path).unwrap(), bytes);
|
|
for name in ["../escape", "a/b", "a\\b", "..", ""] {
|
|
let response = client
|
|
.post(&endpoint)
|
|
.query(&[("name", name)])
|
|
.body("text")
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), 400, "{name}");
|
|
}
|
|
let response = client
|
|
.post(&endpoint)
|
|
.query(&[("name", "large.bin")])
|
|
.body(vec![0; 20 * 1024 * 1024 + 1])
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), 413);
|
|
let response = client
|
|
.post(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{}/structured/attachments", Uuid::new_v4()),
|
|
))
|
|
.query(&[("name", "file.txt")])
|
|
.body("text")
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), 404);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn structured_prompt_queues_during_active_turn_and_cancel_clears_it() {
|
|
let h = spawn("naps-laptop").await;
|
|
let fake = FakeAcp::new();
|
|
let mut env = fake.env();
|
|
env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "200".into());
|
|
let id = structured_session_with_env(&h, env).await;
|
|
let client = reqwest::Client::new();
|
|
let first = client
|
|
.post(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/prompt"),
|
|
))
|
|
.json(&serde_json::json!({"prompt": "first"}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
|
|
// A second prompt while a turn is active is accepted into the queue.
|
|
let second = client
|
|
.post(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/prompt"),
|
|
))
|
|
.json(&serde_json::json!({"prompt": "second"}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(second.status(), reqwest::StatusCode::ACCEPTED);
|
|
|
|
let cancel = client
|
|
.post(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/cancel"),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(cancel.status(), reqwest::StatusCode::NO_CONTENT);
|
|
// After cancel + settlement, the queued "second" prompt drains. Wait for
|
|
// both the first and the drained prompt to reach ACP.
|
|
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("first and drained queued prompt both 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();
|
|
wait_for_session_activity(&h, id, ActivityState::AwaitingInput).await;
|
|
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");
|
|
wait_for_session_activity(&h, id, ActivityState::Idle).await;
|
|
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]
|
|
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_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"}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
|
|
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");
|
|
wait_for_session_activity(&h, id, ActivityState::Error).await;
|
|
let second = client
|
|
.post(url(h.addr, &path))
|
|
.json(&serde_json::json!({"prompt":"available"}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
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,
|
|
auto_approve: false,
|
|
})
|
|
.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]
|
|
async fn structured_replay_honors_cursors_and_limit() {
|
|
let h = spawn("naps-laptop").await;
|
|
let fake = FakeAcp::new();
|
|
let id = structured_session(&h, &fake).await;
|
|
let client = reqwest::Client::new();
|
|
|
|
let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
|
|
loop {
|
|
let events: Vec<serde_json::Value> = client
|
|
.get(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/replay?since=0&limit=1"),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
if !events.is_empty() {
|
|
return events;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(events.len(), 1);
|
|
assert_eq!(events[0]["cursor"], 1);
|
|
assert!(events[0].get("notification").is_none());
|
|
|
|
let invalid = client
|
|
.get(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/replay?since=0&before=2"),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(invalid.status(), reqwest::StatusCode::BAD_REQUEST);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn structured_replay_pages_projected_events_in_cursor_order() {
|
|
let h = spawn("naps-laptop").await;
|
|
let fake = FakeAcp::new();
|
|
let id = structured_session(&h, &fake).await;
|
|
let client = reqwest::Client::new();
|
|
unsafe { libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1) };
|
|
let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
|
|
loop {
|
|
let events: Vec<serde_json::Value> = client
|
|
.get(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/replay?before=99&limit=2"),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
if events.len() == 2 {
|
|
return events;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
events
|
|
.iter()
|
|
.map(|event| event["cursor"].as_u64().unwrap())
|
|
.collect::<Vec<_>>(),
|
|
vec![1, 2]
|
|
);
|
|
let boundary: Vec<serde_json::Value> = client
|
|
.get(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/replay?before=2&limit=10"),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(boundary.len(), 1);
|
|
assert_eq!(boundary[0]["cursor"], 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn structured_replay_rejects_unknown_terminal_and_dead_sessions() {
|
|
let h = spawn("naps-laptop").await;
|
|
let client = reqwest::Client::new();
|
|
let unknown = client
|
|
.get(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{}/structured/replay", Uuid::new_v4()),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(unknown.status(), reqwest::StatusCode::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,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let response = client
|
|
.get(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{}/structured/replay", terminal.id),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
|
|
|
|
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 response = client
|
|
.get(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{dead}/structured/replay"),
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
|
|
let error: ApiError = response.json().await.unwrap();
|
|
assert_eq!(error.error.code, "not_structured");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn structured_ws_rejects_a_dead_session() {
|
|
let h = spawn("naps-laptop").await;
|
|
let fake = FakeAcp::new();
|
|
let id = structured_session(&h, &fake).await;
|
|
h.manager.shutdown().await;
|
|
assert_eq!(h.manager.status(id).await, Some(SessionStatus::Dead));
|
|
|
|
let err = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
|
|
.await
|
|
.expect_err("dead structured session must not upgrade");
|
|
let message = err.to_string();
|
|
assert!(
|
|
message.contains("409") || message.to_lowercase().contains("conflict"),
|
|
"{message}"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn info_returns_daemon_identity() {
|
|
let h = spawn("naps-laptop").await;
|
|
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp["hostname"], "naps-laptop");
|
|
assert_eq!(resp["home_dir"], "/home/test");
|
|
assert!(resp["version"].is_string());
|
|
}
|
|
|
|
/// #159: the web client reads its leader off `/api/info` instead of a
|
|
/// hardcoded constant, so the payload has to carry the configured one — not
|
|
/// the default dressed up as one.
|
|
#[tokio::test]
|
|
async fn info_advertises_the_configured_leader() {
|
|
let h = spawn_with_leader(
|
|
"naps-laptop",
|
|
std::sync::Arc::new(maestro_daemon::projects::ProjectScanner::disabled()),
|
|
"ctrl-b",
|
|
)
|
|
.await;
|
|
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp["leader"], "ctrl-b");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn info_reports_the_default_leader_when_unconfigured() {
|
|
let h = spawn("naps-laptop").await;
|
|
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp["leader"], "ctrl-a");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn info_reports_build_and_instance() {
|
|
// The status indicator needs a release to name and a process to compare
|
|
// against after a reconnect. `commit`/`commit_date` are empty when the
|
|
// binary was built outside a checkout, so only their presence is asserted;
|
|
// `instance` is a fresh uuid per daemon process.
|
|
let h = spawn("naps-laptop").await;
|
|
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert!(resp["build"]["commit"].is_string());
|
|
assert!(resp["build"]["commit_date"].is_string());
|
|
assert!(resp["build"]["dirty"].is_boolean());
|
|
let instance = resp["instance"].as_str().unwrap();
|
|
assert!(Uuid::parse_str(instance).is_ok(), "instance is a uuid");
|
|
|
|
// Same process, same id — a client only treats a *changed* id as a restart.
|
|
let again: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(again["instance"].as_str().unwrap(), instance);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn list_sessions_starts_empty_and_reflects_creates() {
|
|
let h = spawn("naps-laptop").await;
|
|
|
|
let empty: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert!(empty.is_empty());
|
|
|
|
let req = CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: "/tmp".into(),
|
|
command: "sleep 60".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
};
|
|
let created: Session = reqwest::Client::new()
|
|
.post(url(h.addr, "/api/sessions"))
|
|
.json(&req)
|
|
.send()
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
|
|
let after: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(after.len(), 1);
|
|
assert_eq!(after[0].id, created.id);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn info_advertises_spawn_presets() {
|
|
// #26: the daemon's presets ride the info payload so the spawn form needs
|
|
// no extra round-trip. Built-ins + the implicit custom.
|
|
let h = spawn("naps-laptop").await;
|
|
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
let ids: Vec<&str> = resp["presets"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.map(|p| p["id"].as_str().unwrap())
|
|
.collect();
|
|
assert_eq!(ids, ["claude", "opencode", "zsh", "custom"]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn recent_dirs_returns_empty_without_history() {
|
|
// #26: reads this daemon's spawn history. The test fixture wires no
|
|
// history, so the list is empty but the endpoint is well-formed.
|
|
let h = spawn("naps-laptop").await;
|
|
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/recent-dirs"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp["dirs"].as_array().unwrap().len(), 0);
|
|
assert_eq!(resp["truncated"], false);
|
|
}
|
|
|
|
/// #54: `GET /api/projects` lists every git repo under the scan roots, with
|
|
/// real git checkouts — not mocks.
|
|
#[tokio::test]
|
|
async fn projects_lists_repos_under_the_scan_roots() {
|
|
let tmp = tempfile::tempdir().unwrap();
|
|
for name in ["alpha", "beta"] {
|
|
let repo = tmp.path().join(name);
|
|
std::fs::create_dir_all(&repo).unwrap();
|
|
for args in [
|
|
vec!["init", "-q"],
|
|
vec![
|
|
"-c",
|
|
"user.email=t@t",
|
|
"-c",
|
|
"user.name=t",
|
|
"commit",
|
|
"-qm",
|
|
"init",
|
|
"--allow-empty",
|
|
],
|
|
] {
|
|
let out = std::process::Command::new("git")
|
|
.arg("-C")
|
|
.arg(&repo)
|
|
.args(&args)
|
|
.output()
|
|
.unwrap();
|
|
assert!(out.status.success());
|
|
}
|
|
}
|
|
|
|
let scanner = std::sync::Arc::new(maestro_daemon::projects::ProjectScanner::new(
|
|
vec![tmp.path().to_path_buf()],
|
|
3,
|
|
tmp.path().to_path_buf(),
|
|
));
|
|
let h = spawn_with("naps-laptop", scanner).await;
|
|
|
|
let projects: Vec<maestro_protocol::Project> = reqwest::get(url(h.addr, "/api/projects"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
|
|
let mut names: Vec<&str> = projects.iter().map(|p| p.name.as_str()).collect();
|
|
names.sort();
|
|
assert_eq!(names, ["alpha", "beta"]);
|
|
for p in &projects {
|
|
assert!(!p.is_worktree);
|
|
assert_eq!(p.base_repo, None);
|
|
assert!(p.last_activity.is_some(), "fresh commit ⇒ HEAD mtime known");
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn projects_returns_empty_for_a_disabled_scanner() {
|
|
let h = spawn("naps-laptop").await;
|
|
let projects: Vec<maestro_protocol::Project> = reqwest::get(url(h.addr, "/api/projects"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert!(projects.is_empty());
|
|
}
|
|
|
|
/// #53: every returned handle is one word from the shared list.
|
|
fn assert_valid_names(batch: &maestro_protocol::NameBatch) {
|
|
for name in &batch.names {
|
|
assert!(!name.contains('-'), "handle {name:?} is not one word");
|
|
assert!(
|
|
maestro_names::WORDS.contains(&name.as_str()),
|
|
"{name} not in WORDS"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn names_defaults_to_one_handle() {
|
|
let h = spawn("naps-laptop").await;
|
|
let batch: maestro_protocol::NameBatch = reqwest::get(url(h.addr, "/api/names"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(batch.names.len(), 1);
|
|
assert_valid_names(&batch);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn names_honors_count_and_caps_it() {
|
|
use maestro_names::MAX_BATCH;
|
|
let h = spawn("naps-laptop").await;
|
|
|
|
let batch: maestro_protocol::NameBatch = reqwest::get(url(h.addr, "/api/names?count=5"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(batch.names.len(), 5);
|
|
assert_valid_names(&batch);
|
|
|
|
let batch: maestro_protocol::NameBatch =
|
|
reqwest::get(url(h.addr, &format!("/api/names?count={}", MAX_BATCH * 10)))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(batch.names.len(), MAX_BATCH);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn gitroot_probes_path() {
|
|
// #26: resolves the query against the daemon home dir and walks up for a
|
|
// `.git`. The tempdir itself has no `.git`, so the result must either be
|
|
// null (no repo anywhere above) or — in sandboxed environments where /tmp
|
|
// itself sits inside a repo — a path strictly above the tempdir.
|
|
let h = spawn("naps-laptop").await;
|
|
let tmp = tempfile::tempdir().unwrap();
|
|
let resp: serde_json::Value = reqwest::get(url(
|
|
h.addr,
|
|
&format!(
|
|
"/api/gitroot?path={}",
|
|
urlencoding::encode(&tmp.path().to_string_lossy())
|
|
),
|
|
))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
// On hosts where an ancestor of the temp dir is a repo (e.g. `/tmp`
|
|
// inside a checkout), the root legitimately points there. The test only
|
|
// asserts no root was found *inside* the temp dir.
|
|
match resp.get("root").and_then(|v| v.as_str()) {
|
|
None => {}
|
|
Some(root) => assert!(
|
|
!root.starts_with(tmp.path().to_string_lossy().as_ref()),
|
|
"found a repo root inside the fixture at {root:?}"
|
|
),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn delete_unknown_session_returns_404() {
|
|
let h = spawn("naps-laptop").await;
|
|
let resp = reqwest::Client::new()
|
|
.delete(url(h.addr, &format!("/api/sessions/{}", Uuid::new_v4())))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
|
|
}
|
|
|
|
/// #125: the rename endpoint answers with the updated session, and the name
|
|
/// it stored is the trimmed one.
|
|
#[tokio::test]
|
|
async fn patch_session_renames_it() {
|
|
let h = spawn("naps-laptop").await;
|
|
let session = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: "/tmp".into(),
|
|
command: "sleep 60".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let resp = reqwest::Client::new()
|
|
.patch(url(h.addr, &format!("/api/sessions/{}", session.id)))
|
|
.json(&serde_json::json!({ "name": " crimson-otter " }))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::OK);
|
|
let updated: Session = resp.json().await.unwrap();
|
|
assert_eq!(
|
|
updated.metadata.get("name").and_then(|v| v.as_str()),
|
|
Some("crimson-otter")
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn patch_unknown_session_returns_404() {
|
|
let h = spawn("naps-laptop").await;
|
|
let resp = reqwest::Client::new()
|
|
.patch(url(h.addr, &format!("/api/sessions/{}", Uuid::new_v4())))
|
|
.json(&serde_json::json!({ "name": "nope" }))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
|
|
}
|
|
|
|
/// #145: `POST /.../focus` with `focused: true` over HTTP stamps
|
|
/// `last_focused_at`, observable on the next `GET /api/sessions`.
|
|
#[tokio::test]
|
|
async fn post_focus_stamps_last_focused_at() {
|
|
let h = spawn("naps-laptop").await;
|
|
let session = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: "/tmp".into(),
|
|
command: "sleep 60".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(session.last_focused_at, None);
|
|
|
|
let resp = reqwest::Client::new()
|
|
.post(url(h.addr, &format!("/api/sessions/{}/focus", session.id)))
|
|
.json(&serde_json::json!({ "focused": true }))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT);
|
|
|
|
let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
assert!(sessions[0].last_focused_at.is_some());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn delete_owned_session_kills_it() {
|
|
let h = spawn("naps-laptop").await;
|
|
let session = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: "/tmp".into(),
|
|
command: "sleep 60".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let resp = reqwest::Client::new()
|
|
.delete(url(h.addr, &format!("/api/sessions/{}", session.id)))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT);
|
|
assert!(!h.manager.has(session.id).await);
|
|
}
|
|
|
|
/// Spawn a session through the manager and return its id.
|
|
async fn spawn_via_manager(h: &Harness) -> Uuid {
|
|
h.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: "/tmp".into(),
|
|
command: "sleep 60".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap()
|
|
.id
|
|
}
|
|
|
|
/// The agent activity endpoint on the wire (#105): a report lands in the
|
|
/// session snapshot with its detail.
|
|
#[tokio::test]
|
|
async fn post_activity_reports_error_with_detail() {
|
|
let h = spawn("naps-laptop").await;
|
|
let id = spawn_via_manager(&h).await;
|
|
|
|
let resp = reqwest::Client::new()
|
|
.post(url(h.addr, &format!("/api/sessions/{id}/activity")))
|
|
.json(&serde_json::json!({
|
|
"state": "Error",
|
|
"detail": "cargo build failed"
|
|
}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT);
|
|
|
|
let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
let s = sessions.iter().find(|s| s.id == id).unwrap();
|
|
assert_eq!(s.activity, maestro_protocol::ActivityState::Error);
|
|
assert_eq!(s.activity_detail.as_deref(), Some("cargo build failed"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn post_activity_to_unknown_session_returns_404() {
|
|
let h = spawn("naps-laptop").await;
|
|
let resp = reqwest::Client::new()
|
|
.post(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{}/activity", Uuid::new_v4()),
|
|
))
|
|
.json(&serde_json::json!({ "state": "Idle" }))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
|
|
}
|
|
|
|
/// Unknown state values are rejected rather than coerced (#105).
|
|
#[tokio::test]
|
|
async fn post_activity_rejects_an_unknown_state_value() {
|
|
let h = spawn("naps-laptop").await;
|
|
let id = spawn_via_manager(&h).await;
|
|
|
|
let resp = reqwest::Client::new()
|
|
.post(url(h.addr, &format!("/api/sessions/{id}/activity")))
|
|
.json(&serde_json::json!({ "state": "Teleported" }))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert!(resp.status().is_client_error(), "got {}", resp.status());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn post_activity_caps_an_overlong_detail_server_side() {
|
|
let h = spawn("naps-laptop").await;
|
|
let id = spawn_via_manager(&h).await;
|
|
|
|
let resp = reqwest::Client::new()
|
|
.post(url(h.addr, &format!("/api/sessions/{id}/activity")))
|
|
.json(&serde_json::json!({
|
|
"state": "Error",
|
|
"detail": "x".repeat(10_000)
|
|
}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT);
|
|
|
|
let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
|
.await
|
|
.unwrap()
|
|
.json()
|
|
.await
|
|
.unwrap();
|
|
let s = sessions.iter().find(|s| s.id == id).unwrap();
|
|
assert_eq!(
|
|
s.activity_detail.as_ref().map(|d| d.chars().count()),
|
|
Some(maestro_daemon::MAX_REPORT_DETAIL_CHARS)
|
|
);
|
|
}
|
|
|
|
/// The kill guard on the wire (#79): `DELETE /api/sessions/:id` answers 409
|
|
/// with the structured breakdown, and the session keeps running.
|
|
#[tokio::test]
|
|
async fn delete_a_session_whose_worktree_holds_unsaved_work_returns_409_with_the_breakdown() {
|
|
let h = spawn("naps-laptop").await;
|
|
let repo = tempfile::TempDir::new().unwrap();
|
|
common::init_repo(repo.path());
|
|
let session = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: repo.path().to_path_buf(),
|
|
command: "sleep 60".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: Some(WorktreeSpec {
|
|
base_repo: repo.path().to_path_buf(),
|
|
branch: "feat/guarded".into(),
|
|
create_branch: true,
|
|
}),
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
std::fs::write(
|
|
repo.path()
|
|
.join("worktrees")
|
|
.join("feat")
|
|
.join("guarded")
|
|
.join("README"),
|
|
b"unsaved edits\n",
|
|
)
|
|
.unwrap();
|
|
|
|
let resp = reqwest::Client::new()
|
|
.delete(url(h.addr, &format!("/api/sessions/{}", session.id)))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::CONFLICT);
|
|
let body: ApiError = resp.json().await.unwrap();
|
|
assert_eq!(body.error.code, "kill_guarded");
|
|
let breakdown = body.error.unsaved_work.expect("breakdown on the wire");
|
|
assert_eq!(breakdown.uncommitted_files, 1);
|
|
|
|
assert!(h.manager.has(session.id).await, "the session keeps running");
|
|
}
|
|
|
|
/// `?force=true` on the wire (#81): the refusal is overridden and the session
|
|
/// ends, while the guarded worktree survives on disk.
|
|
#[tokio::test]
|
|
async fn delete_with_force_ends_the_session_and_keeps_the_worktree() {
|
|
let h = spawn("naps-laptop").await;
|
|
let repo = tempfile::TempDir::new().unwrap();
|
|
common::init_repo(repo.path());
|
|
let wt_path = repo
|
|
.path()
|
|
.join("worktrees")
|
|
.join("feat")
|
|
.join("guarded")
|
|
.to_path_buf();
|
|
let session = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: repo.path().to_path_buf(),
|
|
command: "sleep 60".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: Some(WorktreeSpec {
|
|
base_repo: repo.path().to_path_buf(),
|
|
branch: "feat/guarded".into(),
|
|
create_branch: true,
|
|
}),
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
std::fs::write(wt_path.join("README"), b"unsaved edits\n").unwrap();
|
|
|
|
let resp = reqwest::Client::new()
|
|
.delete(url(h.addr, &format!("/api/sessions/{}", session.id)))
|
|
.query(&[("force", "true")])
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT);
|
|
|
|
// The tree would have blocked; force must not have removed it.
|
|
assert!(wt_path.exists(), "a forced kill never removes the worktree");
|
|
assert!(!h.manager.has(session.id).await, "the session is gone");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn attach_ws_streams_pty_bytes() {
|
|
let h = spawn("naps-laptop").await;
|
|
let session = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: "/tmp".into(),
|
|
command: "bash --noprofile --norc".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
let ws_url = format!("ws://{}/api/attach/{}", h.addr, session.id);
|
|
let (mut ws, _): (WsClient, _) = connect_async(ws_url).await.unwrap();
|
|
|
|
// Provoke output. The bash process should echo this back via the PTY.
|
|
ws.send(WsMessage::Binary(b"echo hello-attach\n".to_vec()))
|
|
.await
|
|
.unwrap();
|
|
|
|
let saw_hello = tokio::time::timeout(Duration::from_secs(2), async {
|
|
loop {
|
|
match ws.next().await {
|
|
Some(Ok(WsMessage::Binary(bytes))) => {
|
|
if String::from_utf8_lossy(&bytes).contains("hello-attach") {
|
|
return true;
|
|
}
|
|
}
|
|
Some(Ok(_)) => continue,
|
|
Some(Err(_)) | None => return false,
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.unwrap_or(false);
|
|
assert!(saw_hello, "PTY did not echo through the attach WS");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pty_routes_reject_structured_sessions_as_not_terminal() {
|
|
let h = spawn("naps-laptop").await;
|
|
let fake = FakeAcp::new();
|
|
let id = structured_session(&h, &fake).await;
|
|
|
|
let response = reqwest::Client::new()
|
|
.post(url(h.addr, &format!("/api/sessions/{id}/input")))
|
|
.json(&maestro_protocol::SendInputRequest {
|
|
text: "ignored".into(),
|
|
submit: false,
|
|
})
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
|
|
let error: ApiError = response.json().await.unwrap();
|
|
assert_eq!(error.error.code, "not_terminal");
|
|
assert_eq!(error.error.message, "session is not a terminal session");
|
|
|
|
let error = connect_async(ws_url(h.addr, &format!("/api/attach/{id}")))
|
|
.await
|
|
.expect_err("structured session must not upgrade to a PTY socket");
|
|
let tokio_tungstenite::tungstenite::Error::Http(response) = error else {
|
|
panic!("expected an HTTP handshake rejection, got {error}");
|
|
};
|
|
assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
|
|
let body = response.body().as_deref().expect("ApiError response body");
|
|
let error: ApiError = serde_json::from_slice(body).unwrap();
|
|
assert_eq!(error.error.code, "not_terminal");
|
|
assert_eq!(error.error.message, "session is not a terminal session");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn attach_ws_for_unknown_session_returns_404() {
|
|
let h = spawn("naps-laptop").await;
|
|
let ws_url = format!("ws://{}/api/attach/{}", h.addr, Uuid::new_v4());
|
|
let err = connect_async(ws_url)
|
|
.await
|
|
.expect_err("expected handshake fail");
|
|
let msg = err.to_string();
|
|
assert!(
|
|
msg.contains("404") || msg.to_lowercase().contains("not found"),
|
|
"{msg}"
|
|
);
|
|
}
|
|
|
|
/// `WS /api/events` had no daemon-level test: the Snapshot-then-deltas contract
|
|
/// and the subscribe-before-snapshot ordering were only described in a comment.
|
|
#[tokio::test]
|
|
async fn events_ws_sends_a_snapshot_then_streams_deltas() {
|
|
let h = spawn("konishi").await;
|
|
let cwd = tempfile::TempDir::new().unwrap();
|
|
|
|
let (mut ws, _) = connect_async(ws_url(h.addr, "/api/events"))
|
|
.await
|
|
.expect("events ws");
|
|
|
|
match next_event(&mut ws).await {
|
|
maestro_protocol::CatalogEvent::Snapshot { sessions } => assert!(sessions.is_empty()),
|
|
other => panic!("first frame must be a Snapshot, got {other:?}"),
|
|
}
|
|
|
|
let created = h
|
|
.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: cwd.path().to_path_buf(),
|
|
command: "sleep 30".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.expect("create");
|
|
|
|
match next_event(&mut ws).await {
|
|
maestro_protocol::CatalogEvent::SessionAdded { session } => {
|
|
assert_eq!(session.id, created.id);
|
|
}
|
|
other => panic!("expected SessionAdded, got {other:?}"),
|
|
}
|
|
|
|
h.manager.kill(created.id, false).await.expect("kill");
|
|
|
|
// Activity ticks may interleave; the removal is what we are waiting for.
|
|
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
|
|
loop {
|
|
assert!(
|
|
tokio::time::Instant::now() < deadline,
|
|
"timed out waiting for SessionRemoved"
|
|
);
|
|
if let maestro_protocol::CatalogEvent::SessionRemoved { id } = next_event(&mut ws).await {
|
|
assert_eq!(id, created.id);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn events_ws_streams_structured_activity_without_hook_overrides() {
|
|
let h = spawn("konishi").await;
|
|
let fake = FakeAcp::new();
|
|
let mut env = fake.env();
|
|
env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "500".into());
|
|
let (mut ws, _) = connect_async(ws_url(h.addr, "/api/events"))
|
|
.await
|
|
.expect("events ws");
|
|
assert!(matches!(
|
|
next_event(&mut ws).await,
|
|
maestro_protocol::CatalogEvent::Snapshot { .. }
|
|
));
|
|
|
|
let id = structured_session_with_env(&h, env).await;
|
|
let maestro_protocol::CatalogEvent::SessionAdded { session } = next_event(&mut ws).await else {
|
|
panic!("expected structured SessionAdded")
|
|
};
|
|
assert_eq!(session.activity, ActivityState::Starting);
|
|
// The background handshake completes and sends SessionUpdated → Idle.
|
|
let maestro_protocol::CatalogEvent::SessionUpdated { session } = next_event(&mut ws).await
|
|
else {
|
|
panic!("expected SessionUpdated after handshake")
|
|
};
|
|
assert_eq!(session.activity, ActivityState::Idle);
|
|
|
|
let client = reqwest::Client::new();
|
|
let prompt = client
|
|
.post(url(
|
|
h.addr,
|
|
&format!("/api/sessions/{id}/structured/prompt"),
|
|
))
|
|
.json(&serde_json::json!({"prompt": "hello"}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(prompt.status(), reqwest::StatusCode::NO_CONTENT);
|
|
let report = client
|
|
.post(url(h.addr, &format!("/api/sessions/{id}/activity")))
|
|
.json(&serde_json::json!({"state": "Error"}))
|
|
.send()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(report.status(), reqwest::StatusCode::NO_CONTENT);
|
|
wait_for_session_activity(&h, id, ActivityState::Active).await;
|
|
|
|
let mut activities = Vec::new();
|
|
tokio::time::timeout(Duration::from_secs(2), async {
|
|
loop {
|
|
if let maestro_protocol::CatalogEvent::SessionUpdated { session } =
|
|
next_event(&mut ws).await
|
|
{
|
|
activities.push(session.activity);
|
|
if session.activity == ActivityState::Idle {
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.expect("structured turn returns to idle");
|
|
assert!(
|
|
activities.contains(&ActivityState::Active),
|
|
"{activities:?}"
|
|
);
|
|
assert!(
|
|
!activities.contains(&ActivityState::Error),
|
|
"{activities:?}"
|
|
);
|
|
|
|
h.manager.kill(id, false).await.unwrap();
|
|
}
|
|
|
|
/// Each subscriber has its own lag domain now that the publisher task is gone:
|
|
/// two clients connected at different times both get their own Snapshot.
|
|
#[tokio::test]
|
|
async fn every_events_subscriber_gets_its_own_snapshot() {
|
|
let h = spawn("konishi").await;
|
|
let cwd = tempfile::TempDir::new().unwrap();
|
|
|
|
let (mut first, _) = connect_async(ws_url(h.addr, "/api/events")).await.unwrap();
|
|
let _ = next_event(&mut first).await;
|
|
|
|
h.manager
|
|
.create(CreateSessionRequest {
|
|
view: Default::default(),
|
|
cwd: cwd.path().to_path_buf(),
|
|
command: "sleep 30".into(),
|
|
env: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
worktree: None,
|
|
auto_approve: false,
|
|
})
|
|
.await
|
|
.expect("create");
|
|
|
|
let (mut second, _) = connect_async(ws_url(h.addr, "/api/events")).await.unwrap();
|
|
match next_event(&mut second).await {
|
|
maestro_protocol::CatalogEvent::Snapshot { sessions } => {
|
|
assert_eq!(
|
|
sessions.len(),
|
|
1,
|
|
"a late subscriber must see current state"
|
|
);
|
|
}
|
|
other => panic!("expected Snapshot, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
async fn next_event(ws: &mut WsClient) -> maestro_protocol::CatalogEvent {
|
|
loop {
|
|
match tokio::time::timeout(Duration::from_secs(5), ws.next())
|
|
.await
|
|
.expect("events ws timed out")
|
|
{
|
|
Some(Ok(WsMessage::Text(t))) => {
|
|
return serde_json::from_str(&t).expect("decode catalog event")
|
|
}
|
|
Some(Ok(_)) => continue,
|
|
other => panic!("unexpected events frame: {other:?}"),
|
|
}
|
|
}
|
|
}
|