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.
|
/// turn precede this marker on the wire, and therefore in this channel.
|
||||||
/// The event pump must fire `done` after persisting every preceding
|
/// The event pump must fire `done` after persisting every preceding
|
||||||
/// notification.
|
/// 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 {
|
TurnDone {
|
||||||
turn: u64,
|
turn: u64,
|
||||||
result: Result<(), String>,
|
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.
|
/// ACP adapter selected from the structured session's resolved command.
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub enum AcpAdapter {
|
pub enum AcpAdapter {
|
||||||
@@ -301,7 +316,7 @@ enum Command {
|
|||||||
Prompt {
|
Prompt {
|
||||||
text: String,
|
text: String,
|
||||||
turn: u64,
|
turn: u64,
|
||||||
done: oneshot::Sender<Result<(), String>>,
|
done: oneshot::Sender<TurnDonePayload>,
|
||||||
},
|
},
|
||||||
SetConfig {
|
SetConfig {
|
||||||
config_id: String,
|
config_id: String,
|
||||||
@@ -435,7 +450,7 @@ impl AcpSession {
|
|||||||
&self,
|
&self,
|
||||||
text: String,
|
text: String,
|
||||||
turn: u64,
|
turn: u64,
|
||||||
) -> Result<oneshot::Receiver<Result<(), String>>, String> {
|
) -> Result<oneshot::Receiver<TurnDonePayload>, 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 {
|
||||||
@@ -999,7 +1014,11 @@ async fn run_connection<T>(
|
|||||||
.expect("permission turn state")
|
.expect("permission turn state")
|
||||||
.finish(turn);
|
.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 (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);
|
stream.dispatch.queued_turn = Some(turn);
|
||||||
let manager = self.clone();
|
let manager = self.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let lifecycle = match done.await {
|
let (lifecycle, settled) = match done.await {
|
||||||
Ok(Ok(())) => TurnLifecycle::Finish,
|
Ok(payload) => {
|
||||||
Ok(Err(_)) | Err(_) => TurnLifecycle::Failure,
|
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;
|
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);
|
drop(stream);
|
||||||
sync_structured_activity(&self.inner, id).await;
|
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
|
// Every notification for this turn has been persisted and
|
||||||
// broadcast above. Signal the done oneshot so
|
// broadcast above. Signal the done oneshot so
|
||||||
// finish_queued_acp_turn runs with the transcript complete.
|
// finish_queued_acp_turn runs with the transcript complete,
|
||||||
let _ = done.send(result);
|
// 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> {
|
async function jsonOrError<T>(res: Response): Promise<T> {
|
||||||
if (res.ok) {
|
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;
|
return (await res.json()) as T;
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
|
|||||||
Reference in New Issue
Block a user