From faa7f982af274dde15d37bc6ec4ea4aeefc8b4b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 5 Jul 2026 19:49:00 +0200 Subject: [PATCH] feat(web_ui): defer operator /compact to turn end, run when idle too --- hive-ag3nt/src/bin/hive.rs | 18 ++++++++---- hive-ag3nt/src/events.rs | 18 ++++++++++++ hive-ag3nt/src/mcp.rs | 10 ++----- hive-ag3nt/src/turn.rs | 57 ++++++++++++++++++++++++-------------- hive-ag3nt/src/web_ui.rs | 53 ++++++----------------------------- 5 files changed, 77 insertions(+), 79 deletions(-) diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 7248fa4d..5e8f38ca 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -454,13 +454,10 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { login_state.clone(), bus.clone(), socket.to_path_buf(), - files.clone(), - turn_lock.clone(), ); tokio::spawn(async move { - let (label, port, login_state, bus, socket, files, turn_lock) = web_ui_args; - if let Err(e) = web_ui::serve(label, port, login_state, bus, socket, files, turn_lock).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"); } }); @@ -517,7 +514,16 @@ async fn serve_loop( None => match S::recv_next(socket).await { RecvOutcome::Message(first) => first, RecvOutcome::Empty => { - tokio::time::sleep(interval).await; + // Idle: no message this poll. Service a queued operator + // `/compact` here so it runs even when no turn is driving + // (the in-flight case is handled at the end of drive_turn). + let compacted = { + let _guard = turn_lock.lock().await; + turn::run_pending_compact(files, &bus).await + }; + if !compacted { + tokio::time::sleep(interval).await; + } continue; } RecvOutcome::TransportError => { diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 6d50253a..a433ae91 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -688,6 +688,10 @@ pub struct Bus { /// serialized by the serve loop, so the turn boundary is the only point /// where no session file is open. session_reset_pending: Arc, + /// One-shot: run `/compact` after the next turn ends. Consumed at the end + /// of the current/next turn by `turn::drive_turn`. Deferring to the turn + /// boundary keeps compaction from racing a live claude process mid-turn. + compact_pending: Arc, /// Current fresh-claude-session id (FK to `sessions.id`). Set by the /// bin loop after minting a session row on a fresh start; stamped onto /// every `turn_stats` row until the next fresh session. `None` before @@ -777,6 +781,7 @@ impl Bus { last_cost_usage: Arc::new(Mutex::new(None)), rate_limited: Arc::new(AtomicBool::new(was_rate_limited)), session_reset_pending: Arc::new(AtomicBool::new(false)), + compact_pending: Arc::new(AtomicBool::new(false)), session_id: Arc::new(Mutex::new(None)), fresh_session: Arc::new(AtomicBool::new(false)), tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())), @@ -816,6 +821,19 @@ impl Bus { self.session_reset_pending.swap(false, Ordering::SeqCst) } + /// Request a compaction after the next turn ends (deferred to the turn + /// boundary). Idempotent. + pub fn request_compact(&self) { + self.compact_pending.store(true, Ordering::SeqCst); + } + + /// Take + clear the compact one-shot. Returns true iff `drive_turn` should + /// compact at the end of this turn. + #[must_use] + pub fn take_compact(&self) -> bool { + self.compact_pending.swap(false, Ordering::SeqCst) + } + /// Mark that the current turn started a fresh claude session. /// `run_claude` calls this when it creates a new titled session. pub fn mark_fresh_session(&self) { diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 95af42f2..f9afd86f 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -2330,8 +2330,10 @@ pub fn render_claude_config(agent_binary: &str, socket: &std::path::Path) -> Str } #[cfg(test)] -mod recv_hint_tests { +mod tests { use super::{IDLE_WAIT_HINT, SocketReply, format_recv}; + use super::{SERVER_NAME, allowed_mcp_tools}; + use hive_sh4re::ToolGroup; #[test] fn empty_recv_after_wait_appends_idle_hint() { @@ -2345,12 +2347,6 @@ mod recv_hint_tests { let out = format_recv(Ok(SocketReply::Messages(vec![])), false); assert_eq!(out, "(empty)"); } -} - -#[cfg(test)] -mod allowed_tools_tests { - use super::{SERVER_NAME, allowed_mcp_tools}; - use hive_sh4re::ToolGroup; fn qualified(tool: &str) -> String { format!("mcp__{SERVER_NAME}__{tool}") diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index ae28251a..be112cf8 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -150,8 +150,9 @@ pub enum TurnOutcome { /// as `result_kind = "compacted"` in turn stats so the stats page can /// distinguish normal turns from turns that triggered a compaction. Compacted, - /// claude saw "Prompt is too long" — the session needs compacting. - /// Run `compact_session()` then retry the same wake-up prompt. + /// claude saw "Prompt is too long" and even a reactive compact + retry + /// (inside [`InfiniteSession::run`]) couldn't bring it back under the + /// window. Rare; the serve loop treats it like `Ok` (acks the turn). PromptTooLong, /// The Anthropic API refused the request due to a rate limit, per-account /// usage cap, or exhausted credit balance. The serve loop should park for @@ -282,7 +283,7 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco }); result = session.run(&config, prompt, &sink).await; } - match result { + let outcome = match result { Ok(progress) => { if progress.created { // Fresh session minted this turn → flag it so the bin loop @@ -299,7 +300,19 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco } } Err(e) => error_to_turn(e), + }; + // Operator `/compact` (`POST /api/compact`) deferred to the turn boundary: + // run it now that the turn is done, so it works mid-turn rather than only + // when the agent is idle. Only on a healthy turn — no point spawning a + // compaction after a rate-limited / auth-failed / crashed one. + if bus.take_compact() && matches!(outcome, TurnOutcome::Ok | TurnOutcome::Compacted) { + bus.emit(LiveEvent::Note { + text: "operator: /compact — running at turn end".into(), + }); + let _ = session.compact(&config, &sink).await; + return TurnOutcome::Compacted; } + outcome } /// Pre-turn auto-reset check. If context is large AND the prompt cache has @@ -377,30 +390,32 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { } } -/// Operator-initiated `/compact` on the durable session (from the web UI). -/// Resume-only via [`InfiniteSession::compact`]: a missing session is a -/// harmless no-op (never mints an empty session just to compact it). Surfaces -/// the result as a Note and the usual `TurnOutcome`. -pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome { +/// Service a pending operator `/compact` (`Bus::request_compact`) while the +/// agent is idle — the serve loop calls this when a `recv` returns no message, +/// so a queued `/compact` runs even when no turn is driving. (The in-flight +/// case is handled at the end of [`drive_turn`].) Resume-only via +/// [`InfiniteSession::compact`]: a missing session is a harmless no-op. Returns +/// `true` if a compaction ran. +pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus) -> bool { + if !bus.take_compact() { + return false; + } bus.emit(LiveEvent::Note { - text: "running /compact on the session".into(), + text: "operator: /compact — running on idle session".into(), }); + bus.set_state(crate::events::TurnState::Compacting); let config = claude_config(bus, files); let sink = BusSink::new(bus); match infinite_session(bus).compact(&config, &sink).await { - Ok(()) => { - bus.emit(LiveEvent::Note { - text: "/compact done".into(), - }); - TurnOutcome::Compacted - } - Err(e) => { - bus.emit(LiveEvent::Note { - text: format!("/compact failed: {e}"), - }); - error_to_turn(e) - } + Ok(()) => bus.emit(LiveEvent::Note { + text: "/compact done".into(), + }), + Err(e) => bus.emit(LiveEvent::Note { + text: format!("/compact failed: {e}"), + }), } + bus.set_state(crate::events::TurnState::Idle); + true } /// The constant session title for this agent. `HIVE_SESSION_TITLE` overrides diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index f083c8bc..bbb9b889 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -30,7 +30,6 @@ use crate::client; use crate::events::Bus; use crate::login::LoginState; use crate::login_session::{LoginSession, drop_if_finished}; -use crate::turn::TurnFiles; /// Deadline for broker-backed fetches on web-UI request paths. The /// page's critical fields (status, turn state, usage) are all @@ -58,13 +57,6 @@ struct AppState { session: Arc>>>, bus: Bus, socket: PathBuf, - /// Same `TurnFiles` the harness's turn loop uses. Shared so - /// `/api/compact` re-uses the exact MCP config / system prompt - /// claude saw on the last regular turn — keeps the session shape - /// identical across compact + normal turns. - files: TurnFiles, - /// Prevents `/api/compact` from racing with an in-flight normal turn. - turn_lock: TurnLock, /// 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, @@ -88,8 +80,6 @@ pub async fn serve( login: LoginStateCell, bus: Bus, socket: PathBuf, - files: TurnFiles, - turn_lock: TurnLock, ) -> Result<()> { let gui_vnc_port = read_gui_vnc_port(); let static_dir: PathBuf = std::env::var_os("HIVE_STATIC_DIR") @@ -111,8 +101,6 @@ pub async fn serve( session: Arc::new(Mutex::new(None)), bus, socket, - files, - turn_lock, gui_vnc_port, }; let app = Router::new() @@ -1042,12 +1030,6 @@ async fn post_login_cancel(State(state): State) -> Response { (axum::http::StatusCode::OK, "ok").into_response() } -/// Operator-initiated session compaction. Spawns `turn::compact_session` -/// in the background — the HTTP handler returns immediately so the -/// async-form spinner can clear. Output (claude's compaction stream, -/// the "/compact done" note) lands in the live event panel like any -/// other turn. If a regular turn is in flight, claude's own session -/// lock will reject this one and we surface the error as a Note. #[derive(Deserialize)] struct ModelForm { model: String, @@ -1102,34 +1084,15 @@ async fn post_set_effort(State(state): State, Form(form): Form) -> Response { - // Clone the Arc before locking so the guard's lifetime is tied to the - // clone (which we can move into the spawn) rather than to `state`. - let lock = state.turn_lock.clone(); - // Reject immediately if a normal turn is in flight — concurrent access - // to the claude session is unsafe and produces garbled output. - let Ok(guard) = lock.try_lock_owned() else { - return error_response( - StatusCode::CONFLICT, - "turn in flight — wait for it to finish before compacting", - ); - }; - let bus = state.bus.clone(); - let files = state.files.clone(); - tokio::spawn(async move { - let _guard = guard; // keep lock alive for the duration of compaction - bus.emit(crate::events::LiveEvent::Note { - text: "operator: /compact — running on persistent session".into(), - }); - bus.set_state(crate::events::TurnState::Compacting); - let outcome = crate::turn::compact_session(&files, &bus).await; - bus.set_state(crate::events::TurnState::Idle); - // Best-effort manual /compact from the operator: compact_session - // already emits a Note per outcome, so we don't need to re-emit - // here — just record any underlying error to the harness log. - if let crate::turn::TurnOutcome::Failed(e) = outcome { - tracing::warn!(error = %format!("{e:#}"), "operator /compact failed"); - } + state.bus.request_compact(); + state.bus.emit(crate::events::LiveEvent::Note { + text: "operator: /compact queued — runs at the end of the current turn".into(), }); (axum::http::StatusCode::OK, "ok").into_response() }