diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 554b8dbd7..a22f27226 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -498,11 +498,6 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot"); let login_state = Arc::new(Mutex::new(initial)); let bus = Bus::new(); - // Set by the web UI's `/api/cancel` on a successful SIGINT, read-and- - // cleared by `handle_turn` before building the next wake prompt — see - // `hive_sh4re::INTERRUPTED_HINT`. Shared between the web server task and - // the serve loop the same way `bus`/`todo_wake` are. - let interrupted = Arc::new(std::sync::atomic::AtomicBool::new(false)); let stats = TurnStats::open_default(); if let Some(s) = &stats { let (ctx, cost) = s.last_usage(); @@ -540,11 +535,10 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { login_state.clone(), bus.clone(), socket.to_path_buf(), - interrupted.clone(), ); tokio::spawn(async move { - let (label, port, login_state, bus, socket, interrupted) = web_ui_args; - if let Err(e) = web_ui::serve(label, port, login_state, bus, socket, interrupted).await { + let (label, port, login_state, bus, socket) = web_ui_args; + if let Err(e) = web_ui::serve(label, port, login_state, bus, socket).await { tracing::error!(error = %e, "web_ui::serve exited with error"); } }); @@ -601,7 +595,6 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { todo_wake, todos_store, reminder_rx, - interrupted, ) .await } @@ -626,7 +619,6 @@ async fn serve_loop( todo_wake: Arc, todos_store: Option>, mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver, - interrupted: Arc, ) -> Result<()> { tracing::info!(socket = %socket.display(), "harness serve"); S::requeue_inflight(socket).await; @@ -733,23 +725,13 @@ async fn serve_loop( files, &session, graceful_stop_message(), - &interrupted, ) .await; S::graceful_stop_complete(socket).await; return Ok(()); } }; - let ctrl = handle_turn::( - socket, - &bus, - stats.as_ref(), - files, - &session, - next, - &interrupted, - ) - .await; + let ctrl = handle_turn::(socket, &bus, stats.as_ref(), files, &session, next).await; if ctrl.auth_failed { *login_state.lock().unwrap() = LoginState::NeedsLogin; login::wait_for_login( @@ -774,7 +756,6 @@ async fn handle_turn( files: &turn::TurnFiles, session: &turn::AgentSession, first: hive_sh4re::DeliveredMessage, - interrupted: &std::sync::atomic::AtomicBool, ) -> TurnControl { let from = first.from; let body = first.body; @@ -792,70 +773,13 @@ async fn handle_turn( let started_at = chrono::Utc::now().timestamp(); let started_instant = std::time::Instant::now(); let model_at_start = bus.model(); - // Read-and-clear: this wake prompt is the one turn that gets to carry - // the "you were interrupted" banner, then the flag resets so a later - // ordinary turn doesn't repeat a stale notice. - let was_interrupted = interrupted.swap(false, std::sync::atomic::Ordering::Relaxed); - let prompt = serve_common::format_wake_prompt( - msg_id, - &from, - &body, - unread, - redelivered, - was_interrupted, - ); + let prompt = serve_common::format_wake_prompt(msg_id, &from, &body, unread, redelivered); let outcome = turn::drive_turn(&prompt, files, bus, session).await; turn::emit_turn_end(bus, &outcome); bus.set_state(TurnState::Idle); if outcome.is_ok() { S::ack_turn(socket).await; } - handle_turn_error_recovery::(&outcome, bus, socket).await; - if let Some(stats) = stats { - // Fresh session this turn → mint a `sessions` row and set its id on - // the bus so this turn (and subsequent ones until the next fresh - // start) stamp `turn_stats.session_id`. Takes the one-shot flag - // `run_claude` set when it suppressed `--continue`. - if bus.take_fresh_session() { - let sid = stats.start_session(started_at, &model_at_start); - bus.set_session_id(sid); - } - let ended_at = chrono::Utc::now().timestamp(); - let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); - let (open_threads, open_reminders) = S::post_turn_counts(socket).await; - let row = serve_common::build_row(serve_common::TurnRowArgs { - started_at, - ended_at, - duration_ms, - model: model_at_start, - wake_from: from.clone(), - outcome: &outcome, - bus, - open_threads_count: open_threads, - open_reminders_count: open_reminders, - }); - stats.record(&row); - } - let pending = S::inbox_unread(socket).await; - if pending > 0 { - tracing::info!(%pending, "pending messages after turn; fetching next"); - } - TurnControl { - auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)), - } -} - -/// The non-happy-path half of `handle_turn`: react to each `TurnError` -/// variant the turn could have failed with (park-and-retry on rate-limit/ -/// stall/auth, requeue-for-a-fresh-turn on prompt-too-long/session-not- -/// found, notify the parent on a hard failure). Split out purely to keep -/// `handle_turn` itself under clippy's line-count lint — no behavior -/// change from when this lived inline. -async fn handle_turn_error_recovery( - outcome: &turn::TurnOutcome, - bus: &Bus, - socket: &Path, -) { if matches!(outcome, Err(turn::TurnError::RateLimited)) { let secs = turn::rate_limit_sleep_secs(); bus.emit_status("rate_limited"); @@ -906,7 +830,39 @@ async fn handle_turn_error_recovery( tracing::warn!("session-not-found; requeueing message for a fresh turn"); S::requeue_inflight(socket).await; } - if let Err(turn::TurnError::Failed(e)) = outcome { + if let Err(turn::TurnError::Failed(e)) = &outcome { S::send_to_parent(socket, format_turn_failure(e)).await; } + if let Some(stats) = stats { + // Fresh session this turn → mint a `sessions` row and set its id on + // the bus so this turn (and subsequent ones until the next fresh + // start) stamp `turn_stats.session_id`. Takes the one-shot flag + // `run_claude` set when it suppressed `--continue`. + if bus.take_fresh_session() { + let sid = stats.start_session(started_at, &model_at_start); + bus.set_session_id(sid); + } + let ended_at = chrono::Utc::now().timestamp(); + let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); + let (open_threads, open_reminders) = S::post_turn_counts(socket).await; + let row = serve_common::build_row(serve_common::TurnRowArgs { + started_at, + ended_at, + duration_ms, + model: model_at_start, + wake_from: from.clone(), + outcome: &outcome, + bus, + open_threads_count: open_threads, + open_reminders_count: open_reminders, + }); + stats.record(&row); + } + let pending = S::inbox_unread(socket).await; + if pending > 0 { + tracing::info!(%pending, "pending messages after turn; fetching next"); + } + TurnControl { + auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)), + } } diff --git a/hive-agent/src/serve_common.rs b/hive-agent/src/serve_common.rs index fda3c738d..778aee51c 100644 --- a/hive-agent/src/serve_common.rs +++ b/hive-agent/src/serve_common.rs @@ -12,11 +12,7 @@ use crate::turn_stats::TurnStatRow; /// system prompt; this is just the wake signal body. `id` is the broker row /// id, rendered as a `[msg #]` marker so the agent can reference it in /// `ack_until`. `unread` is the inbox depth after this message was popped. -/// `redelivered` prepends a "may already be handled" banner; `interrupted` -/// prepends a "previous turn was /cancel'd" banner instead (the two are -/// mutually exclusive in practice — a redelivered message means the harness -/// itself restarted, which also clears the in-memory interrupted flag — so -/// `redelivered` takes priority if both were somehow set). +/// `redelivered` prepends a "may already be handled" banner. #[must_use] pub fn format_wake_prompt( id: i64, @@ -24,12 +20,9 @@ pub fn format_wake_prompt( body: &str, unread: u64, redelivered: bool, - interrupted: bool, ) -> String { let banner = if redelivered { hive_sh4re::REDELIVERY_HINT - } else if interrupted { - hive_sh4re::INTERRUPTED_HINT } else { "" }; diff --git a/hive-agent/src/web_ui/actions.rs b/hive-agent/src/web_ui/actions.rs index c9b2248eb..52cb4e75b 100644 --- a/hive-agent/src/web_ui/actions.rs +++ b/hive-agent/src/web_ui/actions.rs @@ -54,17 +54,7 @@ pub(super) async fn post_send( pub(super) async fn post_cancel_turn(State(state): State) -> Response { let out = super::sigint_claude().await; let note = match out { - Ok(o) if o.status.success() => { - // A process actually got signalled — the *next* turn's wake - // prompt should tell the agent it was cut off mid-work. Only - // set on an actual signal: a `pkill` exit 1 ("no process to - // interrupt") means /cancel raced an already-finished turn, so - // there's nothing to flag as interrupted. - state - .interrupted - .store(true, std::sync::atomic::Ordering::Relaxed); - "operator: /cancel — sent SIGINT to claude".to_owned() - } + Ok(o) if o.status.success() => "operator: /cancel — sent SIGINT to claude".to_owned(), Ok(o) if o.status.code() == Some(1) => { "operator: /cancel — no claude process to interrupt".to_owned() } diff --git a/hive-agent/src/web_ui/mod.rs b/hive-agent/src/web_ui/mod.rs index 63d66fef4..4693c361d 100644 --- a/hive-agent/src/web_ui/mod.rs +++ b/hive-agent/src/web_ui/mod.rs @@ -19,7 +19,6 @@ mod stream; use std::net::SocketAddr; use std::path::{Path, PathBuf}; -use std::sync::atomic::AtomicBool; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; @@ -58,13 +57,6 @@ struct AppState { /// VNC port from the `HIVE_GUI_VNC_PORT` env var at startup. /// `None` when unset (gui not enabled for this agent). gui_vnc_port: Option, - /// Set by `post_cancel_turn` on a successful SIGINT; read-and-cleared - /// by the serve loop's next `handle_turn` to prepend - /// `hive_sh4re::INTERRUPTED_HINT` to that turn's wake prompt. Shared - /// with the serve loop via the same `Arc` (see `serve_main`) — an - /// in-memory flag, not a marker file, since a `/cancel` from a prior - /// process lifetime isn't meaningful once the harness restarts. - interrupted: Arc, } /// Bind the per-container web listener and serve the SPA. @@ -85,7 +77,6 @@ pub async fn serve( login: LoginStateCell, bus: Bus, socket: PathBuf, - interrupted: Arc, ) -> Result<()> { let gui_vnc_port = read_gui_vnc_port(); let static_dir: PathBuf = std::env::var_os("HIVE_STATIC_DIR") @@ -108,7 +99,6 @@ pub async fn serve( bus, socket, gui_vnc_port, - interrupted, }; let app: Router = Router::new() .route("/api/state", get(state::api_state)) diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 2d6a33017..d2226c78b 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -24,15 +24,6 @@ pub const RECV_BATCH_MAX: u32 = 5; /// server (`recv` tool result) so both surfaces phrase it identically. pub const REDELIVERY_HINT: &str = "[redelivered after harness restart — may already be handled]\n"; -/// Banner prepended to a wake prompt when the previous turn was cut off by -/// an explicit operator `/cancel` (SIGINT) rather than ending normally. Set -/// once, read-and-cleared by the next turn's wake-prompt build — see -/// `hive-agent`'s `post_cancel_turn` (sets it) and `handle_turn` (clears -/// it). Lives here for the same reason as `REDELIVERY_HINT`: a single -/// phrasing, not duplicated between call sites. -pub const INTERRUPTED_HINT: &str = "[your previous turn was interrupted by the operator (/cancel) \ - before it finished — check for new messages before resuming prior work]\n"; - /// Shared "(N more message(s) pending …)" advisory appended after both the /// wake prompt body and the `recv` tool result whenever the inbox still has /// queued messages once the current message/batch is popped. Returns an empty