Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b1f6014a6a | ||
|
|
01db6459cb | ||
|
|
d332c43a9f |
@@ -69,13 +69,28 @@ pub enum AcpUpdate {
|
||||
/// turn precede this marker on the wire, and therefore in this channel.
|
||||
/// The event pump must fire `done` after persisting every preceding
|
||||
/// notification.
|
||||
///
|
||||
/// The `done` oneshot carries both the prompt result and a `settled_tx`
|
||||
/// that the receiver fires after the turn marker is persisted. The
|
||||
/// event pump awaits the corresponding `settled_rx` before processing
|
||||
/// further messages so late-arriving notifications cannot leak into the
|
||||
/// next turn.
|
||||
TurnDone {
|
||||
turn: u64,
|
||||
result: Result<(), String>,
|
||||
done: oneshot::Sender<Result<(), String>>,
|
||||
done: oneshot::Sender<TurnDonePayload>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Payload sent through the `done` oneshot inside [`AcpUpdate::TurnDone`].
|
||||
#[derive(Debug)]
|
||||
pub struct TurnDonePayload {
|
||||
pub result: Result<(), String>,
|
||||
/// The receiver fires this after persisting the turn marker and
|
||||
/// dispatching any queued prompt.
|
||||
pub settled: oneshot::Sender<()>,
|
||||
}
|
||||
|
||||
/// ACP adapter selected from the structured session's resolved command.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum AcpAdapter {
|
||||
@@ -301,7 +316,7 @@ enum Command {
|
||||
Prompt {
|
||||
text: String,
|
||||
turn: u64,
|
||||
done: oneshot::Sender<Result<(), String>>,
|
||||
done: oneshot::Sender<TurnDonePayload>,
|
||||
},
|
||||
SetConfig {
|
||||
config_id: String,
|
||||
@@ -435,7 +450,7 @@ impl AcpSession {
|
||||
&self,
|
||||
text: String,
|
||||
turn: u64,
|
||||
) -> Result<oneshot::Receiver<Result<(), String>>, String> {
|
||||
) -> Result<oneshot::Receiver<TurnDonePayload>, String> {
|
||||
let (done_tx, done_rx) = oneshot::channel();
|
||||
self.commands
|
||||
.send(Command::Prompt {
|
||||
@@ -999,7 +1014,11 @@ async fn run_connection<T>(
|
||||
.expect("permission turn state")
|
||||
.finish(turn);
|
||||
if let Some(done) = done.lock().expect("prompt sender").take() {
|
||||
let _ = done.send(Err(error.to_string()));
|
||||
let (settled_tx, _) = oneshot::channel();
|
||||
let _ = done.send(TurnDonePayload {
|
||||
result: Err(error.to_string()),
|
||||
settled: settled_tx,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -869,11 +869,24 @@ impl SessionManager {
|
||||
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,
|
||||
let (lifecycle, settled) = match done.await {
|
||||
Ok(payload) => {
|
||||
let lifecycle = if payload.result.is_ok() {
|
||||
TurnLifecycle::Finish
|
||||
} else {
|
||||
TurnLifecycle::Failure
|
||||
};
|
||||
(lifecycle, Some(payload.settled))
|
||||
}
|
||||
Err(_) => (TurnLifecycle::Failure, None),
|
||||
};
|
||||
manager.finish_queued_acp_turn(id, turn, lifecycle).await;
|
||||
// Tell the event pump it may resume processing
|
||||
// notifications — the turn marker is persisted
|
||||
// and any queued prompt has been dispatched.
|
||||
if let Some(settled) = settled {
|
||||
let _ = settled.send(());
|
||||
}
|
||||
});
|
||||
drop(stream);
|
||||
sync_structured_activity(&self.inner, id).await;
|
||||
@@ -2443,8 +2456,17 @@ fn start_acp_event_pump(
|
||||
} => {
|
||||
// Every notification for this turn has been persisted and
|
||||
// broadcast above. Signal the done oneshot so
|
||||
// finish_queued_acp_turn runs with the transcript complete.
|
||||
let _ = done.send(result);
|
||||
// finish_queued_acp_turn runs with the transcript complete,
|
||||
// then wait for it to persist the turn marker before we
|
||||
// resume — otherwise late-arriving notifications race into
|
||||
// the next turn.
|
||||
let (settled_tx, settled_rx) = tokio::sync::oneshot::channel();
|
||||
let _ = done.send(maestro_acp::TurnDonePayload {
|
||||
result,
|
||||
settled: settled_tx,
|
||||
});
|
||||
// Block the pump until the turn is fully settled.
|
||||
let _ = settled_rx.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -39,7 +39,7 @@ export class ApiRequestError extends Error {
|
||||
|
||||
async function jsonOrError<T>(res: Response): Promise<T> {
|
||||
if (res.ok) {
|
||||
if (res.status === 204) return undefined as unknown as T;
|
||||
if (res.status === 204 || res.status === 202) return undefined as unknown as T;
|
||||
return (await res.json()) as T;
|
||||
}
|
||||
try {
|
||||
|
||||
Reference in New Issue
Block a user