turn errors reach every surface: Error event end-to-end, no more silent empty replies
This commit is contained in:
parent
da410b1e6a
commit
33068d69ff
5 changed files with 49 additions and 4 deletions
|
|
@ -253,11 +253,11 @@ async fn handle_conversation_stream(
|
|||
let server_clone = server.clone();
|
||||
let conv = conversation_id.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = crate::server::turn::run_turn(
|
||||
let turn = tokio::spawn(async move {
|
||||
crate::server::turn::run_turn(
|
||||
server_clone, conv, &be_tx, event_bus,
|
||||
cancel, interject,
|
||||
).await;
|
||||
).await
|
||||
});
|
||||
|
||||
// Drain the turn's BackendEvent stream. run_turn already handles message
|
||||
|
|
@ -268,12 +268,23 @@ async fn handle_conversation_stream(
|
|||
while let Some(result) = be_rx.recv().await {
|
||||
let be = match result {
|
||||
Ok(be) => be,
|
||||
Err(_) => break,
|
||||
Err(e) => {
|
||||
eprintln!("Turn stream error ({conversation_id}): {e:#}");
|
||||
let _ = tx.send(StreamEvent::Error { message: format!("{e:#}") }).await;
|
||||
break;
|
||||
}
|
||||
};
|
||||
if tx.send(StreamEvent::from(be)).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
// The channel closing means run_turn returned; a turn that died before
|
||||
// emitting anything must still reach the surface as an error, not as a
|
||||
// silently-ended stream (an empty reply reads as the agent going mute).
|
||||
if let Ok(Err(e)) = turn.await {
|
||||
eprintln!("Turn failed ({conversation_id}): {e:#}");
|
||||
let _ = tx.send(StreamEvent::Error { message: format!("{e:#}") }).await;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -284,6 +284,10 @@ pub enum StreamEvent {
|
|||
Interstitial { text: String, register: crate::backend::Register },
|
||||
#[serde(rename = "primary_complete")]
|
||||
PrimaryComplete,
|
||||
/// The turn failed in the substrate. Mirrors BackendEvent::Error so a
|
||||
/// dead turn is never a silently-ended stream.
|
||||
#[serde(rename = "error")]
|
||||
Error { message: String },
|
||||
#[serde(rename = "done")]
|
||||
Done,
|
||||
#[serde(rename = "ping")]
|
||||
|
|
@ -315,6 +319,7 @@ impl StreamEvent {
|
|||
StreamEvent::Outfit { .. } => "outfit",
|
||||
StreamEvent::Interstitial { .. } => "interstitial",
|
||||
StreamEvent::PrimaryComplete => "primary_complete",
|
||||
StreamEvent::Error { .. } => "error",
|
||||
StreamEvent::Done => "done",
|
||||
StreamEvent::Ping => "ping",
|
||||
}
|
||||
|
|
@ -360,6 +365,7 @@ impl From<crate::backend::BackendEvent> for StreamEvent {
|
|||
BE::Interstitial { text, register } => Self::Interstitial { text, register },
|
||||
BE::Keepalive => Self::Ping,
|
||||
BE::PrimaryComplete => Self::PrimaryComplete,
|
||||
BE::Error { message } => Self::Error { message },
|
||||
BE::Done => Self::Done,
|
||||
}
|
||||
}
|
||||
|
|
@ -404,6 +410,7 @@ impl From<StreamEvent> for crate::backend::BackendEvent {
|
|||
StreamEvent::Interstitial { text, register } => BE::Interstitial { text, register },
|
||||
StreamEvent::Ping => BE::Keepalive,
|
||||
StreamEvent::PrimaryComplete => BE::PrimaryComplete,
|
||||
StreamEvent::Error { message } => BE::Error { message },
|
||||
StreamEvent::Done => BE::Done,
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -151,6 +151,11 @@ pub enum BackendEvent {
|
|||
/// events. The substrate does not hold the user hostage to N+1: the
|
||||
/// primary yields, the subconscious presses on.
|
||||
PrimaryComplete,
|
||||
/// The turn failed in the substrate (agent load, provider call, tool
|
||||
/// plumbing). Every surface must show this loud — a swallowed turn
|
||||
/// error reads as the agent going silent, which is worse than any
|
||||
/// error text.
|
||||
Error { message: String },
|
||||
/// Stream ended cleanly.
|
||||
Done,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1238,6 +1238,14 @@ async fn run_chat(
|
|||
BackendEvent::Surfacing { source, content, .. } => {
|
||||
if !json { eprintln!("\n [{}] {}", source, content); }
|
||||
}
|
||||
BackendEvent::Error { message } => {
|
||||
if json {
|
||||
let out = serde_json::json!({"agent": agent.name, "error": message});
|
||||
println!("{}", serde_json::to_string_pretty(&out)?);
|
||||
return Ok(());
|
||||
}
|
||||
eprintln!("\n [turn error] {}", message);
|
||||
}
|
||||
BackendEvent::Done => break,
|
||||
_ => {}
|
||||
}
|
||||
|
|
@ -1278,6 +1286,7 @@ async fn run_chat(
|
|||
BackendEvent::Token(t) => { print!("{}", t); std::io::stdout().flush().ok(); }
|
||||
BackendEvent::Reasoning(r) => eprintln!("\n [thinking] {}", r),
|
||||
BackendEvent::Surfacing { source, content, .. } => eprintln!("\n [{}] {}", source, content),
|
||||
BackendEvent::Error { message } => eprintln!("\n [turn error] {}", message),
|
||||
BackendEvent::Done => break,
|
||||
_ => {}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -395,6 +395,19 @@ impl ChatState {
|
|||
}
|
||||
BackendEvent::Keepalive => {
|
||||
}
|
||||
BackendEvent::Error { message } => {
|
||||
// A dead turn must land visibly and release the input —
|
||||
// leaving `busy` set would freeze the pane on a failure.
|
||||
self.finalize_streaming();
|
||||
self.messages.push(ChatMessage::System {
|
||||
text: format!("turn error: {message}"),
|
||||
ts: Instant::now(),
|
||||
});
|
||||
self.busy = false;
|
||||
self.cancel_token = None;
|
||||
self.phase = TurnPhase::Idle;
|
||||
self.tool_calls_this_turn = 0;
|
||||
}
|
||||
BackendEvent::PrimaryComplete => {
|
||||
self.finalize_streaming();
|
||||
self.busy = false;
|
||||
|
|
|
|||
Loading…
Reference in a new issue