Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
24ed1ac002 | ||
|
|
64741a027e | ||
|
|
cdd4213ff5 | ||
|
|
f8973b3802 | ||
|
|
1e2c0f0a38 | ||
|
|
6fa7dc63e4 | ||
|
|
5c0a4c4640 | ||
|
|
8773671afd | ||
|
|
d9815d86c6 | ||
|
|
13009d0f0e |
@@ -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.
|
the two matchers must agree on, and the fixture table they are tested against.
|
||||||
Read it before changing either bar or either matcher.
|
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
|
`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
|
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
|
daemon answers `/api/info`. The SPA proxies `/api/*` back to the daemon, so only
|
||||||
|
|||||||
+100
-11
@@ -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)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
pub enum AcpError {
|
pub enum AcpError {
|
||||||
/// The adapter is not on `PATH`. The message names the install command
|
/// The adapter is not on `PATH`. The message names the install command
|
||||||
@@ -156,6 +165,7 @@ enum Command {
|
|||||||
Cancel,
|
Cancel,
|
||||||
Prompt {
|
Prompt {
|
||||||
text: String,
|
text: String,
|
||||||
|
turn: u64,
|
||||||
done: oneshot::Sender<Result<(), String>>,
|
done: oneshot::Sender<Result<(), String>>,
|
||||||
},
|
},
|
||||||
Close,
|
Close,
|
||||||
@@ -166,6 +176,31 @@ enum ChildCommand {
|
|||||||
Reap(oneshot::Sender<bool>),
|
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.
|
/// Why the handshake did not produce a session id.
|
||||||
enum ReadyError {
|
enum ReadyError {
|
||||||
/// The adapter answered with an error.
|
/// The adapter answered with an error.
|
||||||
@@ -192,6 +227,7 @@ pub struct AcpSession {
|
|||||||
commands: mpsc::UnboundedSender<Command>,
|
commands: mpsc::UnboundedSender<Command>,
|
||||||
child_commands: mpsc::UnboundedSender<ChildCommand>,
|
child_commands: mpsc::UnboundedSender<ChildCommand>,
|
||||||
updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
|
updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
|
||||||
|
permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
|
||||||
exit: watch::Receiver<Option<ExitStatus>>,
|
exit: watch::Receiver<Option<ExitStatus>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -235,6 +271,13 @@ impl AcpSession {
|
|||||||
self.updates.lock().expect("updates mutex").take()
|
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
|
/// Send `session/cancel`. Fire-and-forget: a closed connection means the
|
||||||
/// child is already gone, which is the outcome cancel was after.
|
/// child is already gone, which is the outcome cancel was after.
|
||||||
pub fn cancel(&self) {
|
pub fn cancel(&self) {
|
||||||
@@ -242,11 +285,16 @@ impl AcpSession {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Send a user prompt and receive completion once the agent ends the turn.
|
/// 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();
|
let (done_tx, done_rx) = oneshot::channel();
|
||||||
self.commands
|
self.commands
|
||||||
.send(Command::Prompt {
|
.send(Command::Prompt {
|
||||||
text,
|
text,
|
||||||
|
turn,
|
||||||
done: done_tx,
|
done: done_tx,
|
||||||
})
|
})
|
||||||
.map_err(|_| "ACP connection is closed".to_string())?;
|
.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));
|
tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
|
||||||
|
|
||||||
let (updates_tx, updates_rx) = mpsc::unbounded_channel();
|
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 (commands_tx, commands_rx) = mpsc::unbounded_channel();
|
||||||
let (ready_tx, ready_rx) = oneshot::channel();
|
let (ready_tx, ready_rx) = oneshot::channel();
|
||||||
let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
|
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.cwd.clone(),
|
||||||
config.load_session.clone(),
|
config.load_session.clone(),
|
||||||
updates_tx,
|
updates_tx,
|
||||||
|
permission_tx,
|
||||||
commands_rx,
|
commands_rx,
|
||||||
ready_tx,
|
ready_tx,
|
||||||
));
|
));
|
||||||
@@ -356,6 +406,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
|
|||||||
commands: commands_tx,
|
commands: commands_tx,
|
||||||
child_commands: child_commands_tx,
|
child_commands: child_commands_tx,
|
||||||
updates: Mutex::new(Some(updates_rx)),
|
updates: Mutex::new(Some(updates_rx)),
|
||||||
|
permission_requests: Mutex::new(Some(permission_rx)),
|
||||||
exit: exit_rx,
|
exit: exit_rx,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -467,6 +518,7 @@ async fn run_connection<T>(
|
|||||||
cwd: PathBuf,
|
cwd: PathBuf,
|
||||||
load_session: Option<String>,
|
load_session: Option<String>,
|
||||||
updates_tx: mpsc::UnboundedSender<SessionNotification>,
|
updates_tx: mpsc::UnboundedSender<SessionNotification>,
|
||||||
|
permission_requests: mpsc::UnboundedSender<PermissionRequest>,
|
||||||
mut commands_rx: mpsc::UnboundedReceiver<Command>,
|
mut commands_rx: mpsc::UnboundedReceiver<Command>,
|
||||||
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
|
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
|
||||||
) where
|
) where
|
||||||
@@ -478,6 +530,10 @@ async fn run_connection<T>(
|
|||||||
// history the event log already holds (#135).
|
// history the event log already holds (#135).
|
||||||
let replaying = Arc::new(AtomicBool::new(false));
|
let replaying = Arc::new(AtomicBool::new(false));
|
||||||
let replaying_seen = replaying.clone();
|
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
|
let result = Client
|
||||||
.builder()
|
.builder()
|
||||||
.name("maestro")
|
.name("maestro")
|
||||||
@@ -495,15 +551,30 @@ async fn run_connection<T>(
|
|||||||
)
|
)
|
||||||
.on_receive_request(
|
.on_receive_request(
|
||||||
async move |request: RequestPermissionRequest, responder, _cx| {
|
async move |request: RequestPermissionRequest, responder, _cx| {
|
||||||
// Stub until #139: decline rather than grant something the
|
let turn = permission_turn_seen
|
||||||
// operator never saw.
|
.lock()
|
||||||
tracing::warn!(
|
.expect("permission turn state")
|
||||||
session = %request.session_id,
|
.callback_turn();
|
||||||
"session/request_permission declined: not implemented yet (#139)"
|
let Some(turn) = turn else {
|
||||||
);
|
return responder.respond(RequestPermissionResponse::new(
|
||||||
responder.respond(RequestPermissionResponse::new(
|
RequestPermissionOutcome::Cancelled,
|
||||||
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!(),
|
agent_client_protocol::on_receive_request!(),
|
||||||
)
|
)
|
||||||
@@ -575,26 +646,44 @@ async fn run_connection<T>(
|
|||||||
tokio::select! {
|
tokio::select! {
|
||||||
cmd = commands_rx.recv() => match cmd {
|
cmd = commands_rx.recv() => match cmd {
|
||||||
Some(Command::Cancel) => {
|
Some(Command::Cancel) => {
|
||||||
|
permission_turn
|
||||||
|
.lock()
|
||||||
|
.expect("permission turn state")
|
||||||
|
.cancel_active();
|
||||||
let _ = cx.send_notification(CancelNotification::new(
|
let _ = cx.send_notification(CancelNotification::new(
|
||||||
SessionId::new(session_id.as_str()),
|
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(
|
let request = PromptRequest::new(
|
||||||
SessionId::new(session_id.as_str()),
|
SessionId::new(session_id.as_str()),
|
||||||
vec![ContentBlock::from(text)],
|
vec![ContentBlock::from(text)],
|
||||||
);
|
);
|
||||||
let done = Arc::new(Mutex::new(Some(done)));
|
let done = Arc::new(Mutex::new(Some(done)));
|
||||||
let callback_done = done.clone();
|
let callback_done = done.clone();
|
||||||
|
let callback_turn = permission_turn.clone();
|
||||||
if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
|
if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
|
||||||
let callback_done = callback_done.clone();
|
let callback_done = callback_done.clone();
|
||||||
|
let callback_turn = callback_turn.clone();
|
||||||
async move {
|
async move {
|
||||||
|
callback_turn
|
||||||
|
.lock()
|
||||||
|
.expect("permission turn state")
|
||||||
|
.finish(turn);
|
||||||
if let Some(done) = callback_done.lock().expect("prompt sender").take() {
|
if let Some(done) = callback_done.lock().expect("prompt sender").take() {
|
||||||
let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
|
let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
|
||||||
}
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}) {
|
}) {
|
||||||
|
permission_turn
|
||||||
|
.lock()
|
||||||
|
.expect("permission turn state")
|
||||||
|
.finish(turn);
|
||||||
if let Some(done) = done.lock().expect("prompt sender").take() {
|
if let Some(done) = done.lock().expect("prompt sender").take() {
|
||||||
let _ = done.send(Err(error.to_string()));
|
let _ = done.send(Err(error.to_string()));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ Environment knobs, all optional:
|
|||||||
so the SIGTERM path is exercised)
|
so the SIGTERM path is exercised)
|
||||||
FAKE_ACP_IGNORE_EOF=1 keep running after stdin closes, until signalled
|
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_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_NO_LOAD=1 advertise `loadSession: false`
|
||||||
FAKE_ACP_LOAD_REPLAY=n on `session/load`, replay n `agent_message_chunk`
|
FAKE_ACP_LOAD_REPLAY=n on `session/load`, replay n `agent_message_chunk`
|
||||||
notifications ("replayed <i>") before answering,
|
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=1 answer every `session/prompt` with a JSON-RPC error
|
||||||
FAKE_ACP_PROMPT_FAIL_ONCE=1 answer only the first `session/prompt` with a
|
FAKE_ACP_PROMPT_FAIL_ONCE=1 answer only the first `session/prompt` with a
|
||||||
JSON-RPC error
|
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>",
|
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
|
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()}"
|
session_id = f"fake-{os.getpid()}"
|
||||||
ticks = 0
|
ticks = 0
|
||||||
prompt_failures = 0
|
prompt_failures = 0
|
||||||
|
pending_prompt_id = None
|
||||||
|
pending_permission_id = None
|
||||||
|
delayed_permission = False
|
||||||
|
|
||||||
|
|
||||||
def chunk(text):
|
def chunk(text):
|
||||||
@@ -102,6 +108,12 @@ for raw in sys.stdin:
|
|||||||
method = msg.get("method")
|
method = msg.get("method")
|
||||||
if method is None:
|
if method is None:
|
||||||
# A response to something we sent; nothing to do.
|
# 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
|
continue
|
||||||
log(f"recv {method}")
|
log(f"recv {method}")
|
||||||
rid = msg.get("id")
|
rid = msg.get("id")
|
||||||
@@ -140,9 +152,47 @@ for raw in sys.stdin:
|
|||||||
elif method == "session/cancel":
|
elif method == "session/cancel":
|
||||||
if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
|
if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
|
||||||
send(chunk("final on cancel"))
|
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":
|
if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
elif method == "session/prompt":
|
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)
|
time.sleep(int(os.environ.get("FAKE_ACP_DELAY_PROMPT_MS", "0")) / 1000)
|
||||||
fail_once = os.environ.get("FAKE_ACP_PROMPT_FAIL_ONCE") == "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):
|
if os.environ.get("FAKE_ACP_PROMPT_FAIL") == "1" or (fail_once and prompt_failures == 0):
|
||||||
|
|||||||
@@ -21,13 +21,12 @@ use maestro_config::SpawnPreset;
|
|||||||
use maestro_protocol::{
|
use maestro_protocol::{
|
||||||
routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
|
routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
|
||||||
RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
|
RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
|
||||||
StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
|
StructuredApprovalDecision, StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
|
||||||
};
|
};
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
use crate::projection::TurnLifecycle;
|
use crate::session::{QueueAcpPrompt, SessionManager};
|
||||||
use crate::session::SessionManager;
|
|
||||||
use crate::sidecar::EventCursor;
|
use crate::sidecar::EventCursor;
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
@@ -70,7 +69,11 @@ pub fn router(state: LocalApiState) -> Router {
|
|||||||
.route(routes::patterns::STRUCTURED, any(structured_ws))
|
.route(routes::patterns::STRUCTURED, any(structured_ws))
|
||||||
.route(routes::patterns::STRUCTURED_REPLAY, get(structured_replay))
|
.route(routes::patterns::STRUCTURED_REPLAY, get(structured_replay))
|
||||||
.route(routes::patterns::STRUCTURED_PROMPT, post(structured_prompt))
|
.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")]
|
#[cfg(feature = "haiku-autoname")]
|
||||||
{
|
{
|
||||||
router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
|
router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
|
||||||
@@ -432,45 +435,24 @@ async fn structured_prompt(
|
|||||||
Path(id): Path<Uuid>,
|
Path(id): Path<Uuid>,
|
||||||
Json(body): Json<StructuredPromptRequest>,
|
Json(body): Json<StructuredPromptRequest>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
let (acp, turn) = match s.manager.begin_acp_turn(id).await {
|
match s.manager.queue_acp_prompt(id, body.prompt).await {
|
||||||
Ok(crate::session::BeginAcpTurn::Started(acp, turn)) => (acp, turn),
|
Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(),
|
||||||
Ok(crate::session::BeginAcpTurn::InFlight) => {
|
Ok(QueueAcpPrompt::InFlight) => error_response(
|
||||||
return error_response(
|
StatusCode::CONFLICT,
|
||||||
StatusCode::CONFLICT,
|
"turn_in_flight",
|
||||||
"turn_in_flight",
|
"the structured session already has an active turn",
|
||||||
"the structured session already has an active turn",
|
),
|
||||||
)
|
Ok(QueueAcpPrompt::Unavailable) => error_response(
|
||||||
}
|
StatusCode::CONFLICT,
|
||||||
Ok(crate::session::BeginAcpTurn::Unavailable) => {
|
"not_structured",
|
||||||
return error_response(
|
"session is not a live structured session",
|
||||||
StatusCode::CONFLICT,
|
),
|
||||||
"not_structured",
|
Ok(QueueAcpPrompt::Failed(message)) => {
|
||||||
"session is not a live structured session",
|
error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
|
||||||
)
|
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
let info = e.to_error_info();
|
let info = e.to_error_info();
|
||||||
return error_response(session_error_status(&info.code), &info.code, &info.message);
|
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)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -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
|
/// Convenience constructor — pulls the bits `LocalApiState` needs from
|
||||||
/// `DaemonConfig` so callers don't have to assemble it by hand. The project
|
/// `DaemonConfig` so callers don't have to assemble it by hand. The project
|
||||||
/// scanner starts with its cache cold; wire
|
/// scanner starts with its cache cold; wire
|
||||||
|
|||||||
@@ -155,6 +155,61 @@ impl Projection {
|
|||||||
self.active_turn
|
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(
|
fn content(
|
||||||
&mut self,
|
&mut self,
|
||||||
seq: u64,
|
seq: u64,
|
||||||
@@ -632,4 +687,68 @@ mod tests {
|
|||||||
fn failed_turn_does_not_reactivate_unfinished_tools() {
|
fn failed_turn_does_not_reactivate_unfinished_tools() {
|
||||||
terminal_turn_drops_unfinished_tools(TurnLifecycle::Failure, "failed-1", "next-1");
|
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"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,18 +12,21 @@ use std::sync::Arc;
|
|||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use chrono::Utc;
|
use chrono::Utc;
|
||||||
use maestro_acp::{AcpSession, SpawnConfig};
|
use maestro_acp::{AcpSession, PermissionRequest, SpawnConfig};
|
||||||
use maestro_protocol::{
|
use maestro_protocol::{
|
||||||
ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
|
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 uuid::Uuid;
|
||||||
|
|
||||||
use crate::panes::Panes;
|
use crate::panes::Panes;
|
||||||
use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
|
use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
|
||||||
use crate::pty::{pipe_pane_command, PtyBridge};
|
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::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
|
||||||
use crate::tmux::TmuxError;
|
use crate::tmux::TmuxError;
|
||||||
use crate::viewport::{ViewportId, Viewports};
|
use crate::viewport::{ViewportId, Viewports};
|
||||||
@@ -118,6 +121,33 @@ struct AcpEventStream {
|
|||||||
live: broadcast::Sender<LoggedEvent>,
|
live: broadcast::Sender<LoggedEvent>,
|
||||||
projection: Projection,
|
projection: Projection,
|
||||||
projected_live: broadcast::Sender<ProjectionUpdate>,
|
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)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
@@ -156,6 +186,15 @@ pub enum BeginAcpTurn {
|
|||||||
Unavailable,
|
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 {
|
impl SessionError {
|
||||||
pub fn to_error_info(&self) -> ErrorInfo {
|
pub fn to_error_info(&self) -> ErrorInfo {
|
||||||
let code = match self {
|
let code = match self {
|
||||||
@@ -406,7 +445,7 @@ impl SessionManager {
|
|||||||
let mut projection = Projection::default();
|
let mut projection = Projection::default();
|
||||||
let mut replay = Vec::new();
|
let mut replay = Vec::new();
|
||||||
for event in events {
|
for event in events {
|
||||||
let update = projection.apply(event.seq, &event.notification);
|
let update = apply_persisted_event(&mut projection, &event);
|
||||||
let include = match cursor {
|
let include = match cursor {
|
||||||
EventCursor::Since(since) => event.seq > since,
|
EventCursor::Since(since) => event.seq > since,
|
||||||
EventCursor::Before(Some(before)) => event.seq < before,
|
EventCursor::Before(Some(before)) => event.seq < before,
|
||||||
@@ -488,6 +527,10 @@ impl SessionManager {
|
|||||||
return Ok(false);
|
return Ok(false);
|
||||||
};
|
};
|
||||||
let mut stream = stream.lock().await;
|
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 {
|
let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) else {
|
||||||
return Ok(false);
|
return Ok(false);
|
||||||
};
|
};
|
||||||
@@ -518,6 +561,150 @@ impl SessionManager {
|
|||||||
Ok(BeginAcpTurn::Started(acp, turn))
|
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.
|
/// Cancel an ACP turn and immediately clear its client-facing lifecycle.
|
||||||
pub async fn cancel_acp_turn(&self, id: Uuid) -> Result<bool, SessionError> {
|
pub async fn cancel_acp_turn(&self, id: Uuid) -> Result<bool, SessionError> {
|
||||||
let (acp, stream) = {
|
let (acp, stream) = {
|
||||||
@@ -531,19 +718,93 @@ impl SessionManager {
|
|||||||
let (Some(acp), Some(stream)) = (acp, stream) else {
|
let (Some(acp), Some(stream)) = (acp, stream) else {
|
||||||
return Ok(false);
|
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 {
|
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);
|
return Ok(true);
|
||||||
};
|
};
|
||||||
acp.cancel();
|
|
||||||
let projected = stream
|
let projected = stream
|
||||||
.projection
|
.projection
|
||||||
.turn_lifecycle(TurnLifecycle::Cancel, turn)
|
.turn_lifecycle(TurnLifecycle::Cancel, turn)
|
||||||
.expect("active turn remains current while stream is locked");
|
.expect("active turn remains current while stream is locked");
|
||||||
let _ = stream.projected_live.send(projected);
|
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)
|
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
|
/// End every adapter child before the daemon exits (#134): each gets
|
||||||
/// `session/cancel`, then SIGTERM, concurrently. The sessions stay in
|
/// `session/cancel`, then SIGTERM, concurrently. The sessions stay in
|
||||||
/// the map, marked Dead and written through to their sidecars, so what
|
/// the map, marked Dead and written through to their sidecars, so what
|
||||||
@@ -569,6 +830,9 @@ impl SessionManager {
|
|||||||
if children.is_empty() {
|
if children.is_empty() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
for (id, _, _) in &children {
|
||||||
|
self.cancel_acp_approvals(*id).await;
|
||||||
|
}
|
||||||
let reaped =
|
let reaped =
|
||||||
futures_util::future::join_all(children.iter().map(|(_, acp, _)| acp.reap_if_exited()))
|
futures_util::future::join_all(children.iter().map(|(_, acp, _)| acp.reap_if_exited()))
|
||||||
.await;
|
.await;
|
||||||
@@ -818,13 +1082,16 @@ impl SessionManager {
|
|||||||
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
|
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
|
||||||
let mut projection = Projection::default();
|
let mut projection = Projection::default();
|
||||||
for event in &persisted {
|
for event in &persisted {
|
||||||
projection.apply(event.seq, &event.notification);
|
apply_persisted_event(&mut projection, event);
|
||||||
}
|
}
|
||||||
Some(Arc::new(Mutex::new(AcpEventStream {
|
Some(Arc::new(Mutex::new(AcpEventStream {
|
||||||
log,
|
log,
|
||||||
live: tx,
|
live: tx,
|
||||||
projection,
|
projection,
|
||||||
projected_live: projected_tx,
|
projected_live: projected_tx,
|
||||||
|
pending_approvals: HashMap::new(),
|
||||||
|
operation: Arc::new(Mutex::new(())),
|
||||||
|
dispatch: AcpDispatchState::default(),
|
||||||
})))
|
})))
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
@@ -933,7 +1200,8 @@ impl SessionManager {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
if let Some((acp, events)) = event_pump {
|
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) {
|
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
|
||||||
record.acp_event_pump = Some(done);
|
record.acp_event_pump = Some(done);
|
||||||
}
|
}
|
||||||
@@ -1260,13 +1528,16 @@ impl SessionManager {
|
|||||||
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
|
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
|
||||||
let mut projection = Projection::default();
|
let mut projection = Projection::default();
|
||||||
for event in &persisted {
|
for event in &persisted {
|
||||||
projection.apply(event.seq, &event.notification);
|
apply_persisted_event(&mut projection, event);
|
||||||
}
|
}
|
||||||
Some(Arc::new(Mutex::new(AcpEventStream {
|
Some(Arc::new(Mutex::new(AcpEventStream {
|
||||||
log,
|
log,
|
||||||
live: tx,
|
live: tx,
|
||||||
projection,
|
projection,
|
||||||
projected_live: projected_tx,
|
projected_live: projected_tx,
|
||||||
|
pending_approvals: HashMap::new(),
|
||||||
|
operation: Arc::new(Mutex::new(())),
|
||||||
|
dispatch: AcpDispatchState::default(),
|
||||||
})))
|
})))
|
||||||
}
|
}
|
||||||
Err(error) => {
|
Err(error) => {
|
||||||
@@ -1318,7 +1589,8 @@ impl SessionManager {
|
|||||||
},
|
},
|
||||||
);
|
);
|
||||||
if let Some((acp, events)) = event_pump {
|
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) {
|
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
|
||||||
record.acp_event_pump = Some(done);
|
record.acp_event_pump = Some(done);
|
||||||
}
|
}
|
||||||
@@ -1672,10 +1944,15 @@ fn start_acp_event_pump(
|
|||||||
while let Some(notification) = updates.recv().await {
|
while let Some(notification) = updates.recv().await {
|
||||||
let append_error = {
|
let append_error = {
|
||||||
let mut events = events.lock().await;
|
let mut events = events.lock().await;
|
||||||
match events.log.append(¬ification) {
|
let event = PersistedEvent::Notification(Box::new(notification));
|
||||||
|
match events.log.append(&event) {
|
||||||
Ok(seq) => {
|
Ok(seq) => {
|
||||||
let projected = events.projection.apply(seq, ¬ification);
|
let logged = LoggedEvent {
|
||||||
let _ = events.live.send(LoggedEvent { seq, notification });
|
seq,
|
||||||
|
event: event.clone(),
|
||||||
|
};
|
||||||
|
let projected = apply_persisted_event(&mut events.projection, &logged);
|
||||||
|
let _ = events.live.send(logged);
|
||||||
let _ = events.projected_live.send(projected);
|
let _ = events.projected_live.send(projected);
|
||||||
None
|
None
|
||||||
}
|
}
|
||||||
@@ -1699,6 +1976,112 @@ fn start_acp_event_pump(
|
|||||||
done_rx
|
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>) {
|
async fn wait_for_event_pump(mut done: watch::Receiver<bool>) {
|
||||||
if !*done.borrow() {
|
if !*done.borrow() {
|
||||||
let _ = done.wait_for(|finished| *finished).await;
|
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 {
|
let Some((mut session, worktree)) = flipped else {
|
||||||
return false;
|
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");
|
tracing::info!(%id, "session process gone — marking dead");
|
||||||
|
|
||||||
if settle_worktree {
|
if settle_worktree {
|
||||||
@@ -2130,6 +2522,9 @@ async fn cleanup_record(
|
|||||||
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
||||||
}
|
}
|
||||||
if let Some(acp) = &record.acp {
|
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).
|
// `session/cancel`, a short wait, then SIGTERM (#134).
|
||||||
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
|
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
|
||||||
.await;
|
.await;
|
||||||
@@ -2163,3 +2558,58 @@ async fn cleanup_record(
|
|||||||
}
|
}
|
||||||
let _ = inner.events.send(CatalogEvent::SessionRemoved { id });
|
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
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -7,8 +7,8 @@
|
|||||||
//! advertised but with empty metadata (handled by the caller, not here).
|
//! advertised but with empty metadata (handled by the caller, not here).
|
||||||
//!
|
//!
|
||||||
//! A structured session (#135) also owns `<id>.events.jsonl` next to its
|
//! A structured session (#135) also owns `<id>.events.jsonl` next to its
|
||||||
//! sidecar: every `session/update` notification its adapter sent, one JSON
|
//! sidecar: every `session/update` notification and daemon approval row is one
|
||||||
//! object per line, appended before it is broadcast. The log is the source of
|
//! 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
|
//! 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
|
//! 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
|
//! 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 std::path::{Path, PathBuf};
|
||||||
|
|
||||||
use maestro_acp::schema::SessionNotification;
|
use maestro_acp::schema::SessionNotification;
|
||||||
use maestro_protocol::Session;
|
use maestro_protocol::{Session, TranscriptRow};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
#[derive(Debug, thiserror::Error)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
@@ -150,10 +150,10 @@ impl SidecarStore {
|
|||||||
}
|
}
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
let Ok(notification) = serde_json::from_slice::<SessionNotification>(&line) else {
|
let Ok(event) = PersistedEvent::decode(&line) else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
out.push_back(LoggedEvent { seq, notification });
|
out.push_back(LoggedEvent { seq, event });
|
||||||
match cursor {
|
match cursor {
|
||||||
EventCursor::Since(_) if out.len() >= limit => break,
|
EventCursor::Since(_) if out.len() >= limit => break,
|
||||||
EventCursor::Before(_) if out.len() > limit => {
|
EventCursor::Before(_) if out.len() > limit => {
|
||||||
@@ -190,6 +190,7 @@ impl SidecarStore {
|
|||||||
/// whichever does not exist.
|
/// whichever does not exist.
|
||||||
pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
|
pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
|
||||||
remove_if_present(&self.events_path_for(id))?;
|
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.recovery_path_for(id))?;
|
||||||
remove_if_present(&self.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.
|
/// One line of a session's event log.
|
||||||
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
|
#[derive(Debug, Clone, PartialEq)]
|
||||||
pub struct LoggedEvent {
|
pub struct LoggedEvent {
|
||||||
/// 1-based line position in the log; the cursor for [`EventCursor`].
|
/// 1-based line position in the log; the cursor for [`EventCursor`].
|
||||||
pub seq: u64,
|
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
|
/// 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
|
/// handed to the kernel in a single write before this returns, so a
|
||||||
/// caller that broadcasts afterwards never announces an event the log
|
/// caller that broadcasts afterwards never announces an event the log
|
||||||
/// would not replay — short of the machine losing power.
|
/// would not replay — short of the machine losing power.
|
||||||
pub fn append(&mut self, notification: &SessionNotification) -> Result<u64, SidecarError> {
|
pub fn append(&mut self, event: impl Into<PersistedEvent>) -> Result<u64, SidecarError> {
|
||||||
let mut line = serde_json::to_vec(notification)?;
|
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');
|
line.push(b'\n');
|
||||||
self.file.write_all(&line)?;
|
self.file.write_all(&line)?;
|
||||||
self.file.flush()?;
|
self.file.flush()?;
|
||||||
|
|||||||
@@ -83,7 +83,10 @@ impl Fake {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
|
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 ¬ification.update {
|
||||||
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
|
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
|
||||||
ContentBlock::Text(text) => &text.text,
|
ContentBlock::Text(text) => &text.text,
|
||||||
other => panic!("unexpected content {other:?}"),
|
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();
|
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]
|
#[tokio::test]
|
||||||
async fn recovery_loads_the_acp_session_without_duplicating_history() {
|
async fn recovery_loads_the_acp_session_without_duplicating_history() {
|
||||||
let fx = DaemonFixture::new();
|
let fx = DaemonFixture::new();
|
||||||
|
|||||||
@@ -142,6 +142,10 @@ impl FakeAcp {
|
|||||||
.next_back()
|
.next_back()
|
||||||
.expect("fake ACP started")
|
.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 {
|
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
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT);
|
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]
|
#[tokio::test]
|
||||||
|
|||||||
@@ -6,8 +6,8 @@ use std::collections::HashMap;
|
|||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
|
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
use maestro_daemon::sidecar::SidecarStore;
|
use maestro_daemon::sidecar::{PersistedEvent, SidecarStore};
|
||||||
use maestro_protocol::{Session, SessionStatus, SessionView};
|
use maestro_protocol::{Session, SessionStatus, SessionView, TranscriptRow, TranscriptRowKind};
|
||||||
use serde_json::Value;
|
use serde_json::Value;
|
||||||
use tempfile::TempDir;
|
use tempfile::TempDir;
|
||||||
use uuid::Uuid;
|
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]
|
#[test]
|
||||||
fn append_assigns_line_position_as_seq_and_reads_back() {
|
fn append_assigns_line_position_as_seq_and_reads_back() {
|
||||||
let dir = TempDir::new().unwrap();
|
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");
|
let mut log = store.open_events(id).expect("open");
|
||||||
assert_eq!(log.last_seq(), 0);
|
assert_eq!(log.last_seq(), 0);
|
||||||
assert_eq!(log.append(¬e("a")).unwrap(), 1);
|
assert_eq!(log.append(note("a")).unwrap(), 1);
|
||||||
assert_eq!(log.append(¬e("b")).unwrap(), 2);
|
assert_eq!(log.append(note("b")).unwrap(), 2);
|
||||||
assert_eq!(log.append(¬e("c")).unwrap(), 3);
|
assert_eq!(log.append(note("c")).unwrap(), 3);
|
||||||
assert_eq!(log.last_seq(), 3);
|
assert_eq!(log.last_seq(), 3);
|
||||||
|
|
||||||
let path = store.events_path_for(id);
|
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();
|
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| e.seq).collect::<Vec<_>>(), vec![1, 2, 3]);
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
all.iter()
|
all.iter().map(event_text).collect::<Vec<_>>(),
|
||||||
.map(|e| text_of(&e.notification))
|
|
||||||
.collect::<Vec<_>>(),
|
|
||||||
vec!["a", "b", "c"]
|
vec!["a", "b", "c"]
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
@@ -225,7 +230,7 @@ fn since_reads_forward_and_before_reads_backward_with_limit() {
|
|||||||
let id = Uuid::new_v4();
|
let id = Uuid::new_v4();
|
||||||
let mut log = store.open_events(id).unwrap();
|
let mut log = store.open_events(id).unwrap();
|
||||||
for i in 1..=6 {
|
for i in 1..=6 {
|
||||||
log.append(¬e(&format!("n{i}"))).unwrap();
|
log.append(note(&format!("n{i}"))).unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
// `since` is exclusive: what a client that already holds seq 2 needs.
|
// `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 store = SidecarStore::new(dir.path());
|
||||||
let id = Uuid::new_v4();
|
let id = Uuid::new_v4();
|
||||||
let mut log = store.open_events(id).unwrap();
|
let mut log = store.open_events(id).unwrap();
|
||||||
log.append(¬e("first")).unwrap();
|
log.append(note("first")).unwrap();
|
||||||
drop(log);
|
drop(log);
|
||||||
|
|
||||||
// Garbage in the middle and a line that is JSON but not a notification.
|
// 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();
|
std::fs::write(&path, raw).unwrap();
|
||||||
let mut log = store.open_events(id).unwrap();
|
let mut log = store.open_events(id).unwrap();
|
||||||
assert_eq!(log.last_seq(), 3);
|
assert_eq!(log.last_seq(), 3);
|
||||||
assert_eq!(log.append(¬e("fourth")).unwrap(), 4);
|
assert_eq!(log.append(note("fourth")).unwrap(), 4);
|
||||||
|
|
||||||
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
|
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!(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]
|
#[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 store = SidecarStore::new(dir.path());
|
||||||
let id = Uuid::new_v4();
|
let id = Uuid::new_v4();
|
||||||
let mut log = store.open_events(id).unwrap();
|
let mut log = store.open_events(id).unwrap();
|
||||||
log.append(¬e("whole")).unwrap();
|
log.append(note("whole")).unwrap();
|
||||||
drop(log);
|
drop(log);
|
||||||
|
|
||||||
let path = store.events_path_for(id);
|
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.
|
// glue onto it; the partial keeps a seq of its own.
|
||||||
let mut log = store.open_events(id).unwrap();
|
let mut log = store.open_events(id).unwrap();
|
||||||
assert_eq!(log.last_seq(), 2);
|
assert_eq!(log.last_seq(), 2);
|
||||||
assert_eq!(log.append(¬e("after")).unwrap(), 3);
|
assert_eq!(log.append(note("after")).unwrap(), 3);
|
||||||
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
|
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!(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();
|
let raw = std::fs::read_to_string(&path).unwrap();
|
||||||
assert_eq!(raw.lines().count(), 3, "{raw}");
|
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();
|
let id = Uuid::new_v4();
|
||||||
store.write(&sample(id)).unwrap();
|
store.write(&sample(id)).unwrap();
|
||||||
let mut log = store.open_events(id).unwrap();
|
let mut log = store.open_events(id).unwrap();
|
||||||
log.append(¬e("x")).unwrap();
|
log.append(note("x")).unwrap();
|
||||||
drop(log);
|
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());
|
assert!(store.events_path_for(id).exists());
|
||||||
|
|
||||||
let loaded = store.load_all().unwrap();
|
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();
|
store.delete(id).unwrap();
|
||||||
assert!(!store.path_for(id).exists());
|
assert!(!store.path_for(id).exists());
|
||||||
assert!(!store.events_path_for(id).exists());
|
assert!(!store.events_path_for(id).exists());
|
||||||
|
assert!(!legacy_approvals.exists());
|
||||||
// Idempotent with the log already gone.
|
// Idempotent with the log already gone.
|
||||||
store.delete(id).unwrap();
|
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(¬e("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);
|
||||||
|
}
|
||||||
|
|||||||
@@ -872,6 +872,15 @@ pub struct StructuredPromptRequest {
|
|||||||
pub prompt: String,
|
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)
|
// Client ↔ Daemon control messages (text frames on the attach WS)
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -91,6 +91,11 @@ pub fn structured_cancel(id: Uuid) -> String {
|
|||||||
format!("{SESSIONS}/{id}/structured/cancel")
|
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
|
/// Axum-style patterns for the two parameterised routes, so the server
|
||||||
/// registers the same paths [`session`] and [`attach`] build.
|
/// registers the same paths [`session`] and [`attach`] build.
|
||||||
pub mod patterns {
|
pub mod patterns {
|
||||||
@@ -104,6 +109,7 @@ pub mod patterns {
|
|||||||
pub const STRUCTURED_REPLAY: &str = "/api/sessions/:id/structured/replay";
|
pub const STRUCTURED_REPLAY: &str = "/api/sessions/:id/structured/replay";
|
||||||
pub const STRUCTURED_PROMPT: &str = "/api/sessions/:id/structured/prompt";
|
pub const STRUCTURED_PROMPT: &str = "/api/sessions/:id/structured/prompt";
|
||||||
pub const STRUCTURED_CANCEL: &str = "/api/sessions/:id/structured/cancel";
|
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.
|
/// A daemon base URL that could not be turned into a WebSocket base.
|
||||||
|
|||||||
+15
-5
@@ -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
|
**Updating is the same command.** Re-running `./install.sh` detects an existing
|
||||||
systemd unit, refreshes it, and runs `systemctl --user restart maestro` instead
|
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
|
`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
|
every tmux-backed session) survives, and the new daemon re-adopts them on
|
||||||
written before this default lacks the line; re-run `./install.sh` to refresh it.
|
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 …`
|
Inside the script, `cargo install --git https://git.naps.pt/naps62/maestro …`
|
||||||
uses your already-configured git credentials for this host (HTTPS helper or
|
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,
|
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
|
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
|
## 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
|
itself. A daemon restart ends an in-flight structured turn, but preserves its
|
||||||
worktree, session record, and event-log transcript. The replacement daemon
|
worktree, session record, and event-log transcript. The replacement daemon
|
||||||
starts a new adapter, resumes it with `session/load`, and replays the
|
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
|
## The browser UI
|
||||||
|
|
||||||
|
|||||||
+39
-4
@@ -2,10 +2,16 @@
|
|||||||
|
|
||||||
Maestro is upgraded in place: the operator reinstalls the binary and systemd
|
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.
|
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
|
tmux keeps **terminal** sessions alive across that (see `docs/INSTALL.md`),
|
||||||
lose their sockets. This document is the contract for what a client does about
|
but clients lose their sockets either way. This document is the contract for
|
||||||
it — what the daemon publishes, how a client tells a restart from a blip, and
|
what a client does about it — what the daemon publishes, how a client tells a
|
||||||
what the user sees while it happens.
|
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
|
## 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
|
moment a probe succeeds — there is no "retry" button, because the client is
|
||||||
already retrying.
|
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
|
## Timing
|
||||||
|
|
||||||
- **Grace window** — 1200 ms between the socket dropping and the overlay
|
- **Grace window** — 1200 ms between the socket dropping and the overlay
|
||||||
|
|||||||
@@ -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.
|
||||||
Reference in New Issue
Block a user