vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(web_ui): defer operator /compact to turn end, run when idle too

Browse source
This commit is contained in:
müde 2026-07-05 19:49:00 +02:00
commit faa7f982af
5 changed files with 77 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

18
hive-ag3nt/src/bin/hive.rs
View file

@ -454,13 +454,10 @@ async fn serve_main<S: Surface>(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<S: Surface>(
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 => {

18
hive-ag3nt/src/events.rs
View file

@ -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<AtomicBool>,
/// 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<AtomicBool>,
/// 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) {

10
hive-ag3nt/src/mcp.rs
View file

@ -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}")

57
hive-ag3nt/src/turn.rs
View file

@ -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

53
hive-ag3nt/src/web_ui.rs
View file

@ -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<Mutex<Option<Arc<LoginSession>>>>,
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<u16>,
@ -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<AppState>) -> 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<AppState>, Form(form): Form<EffortF
(axum::http::StatusCode::OK, "ok").into_response()
}
/// Operator-initiated `/compact`. Deferred: sets the `compact_pending` flag
/// that `turn::drive_turn` consumes at the end of the current/next turn, so it
/// works while a turn is in flight (a mid-turn compaction would race the live
/// claude process) rather than only when the agent is idle. Returns 200
/// immediately; the compaction stream lands in the live panel when it runs.
async fn post_compact(State(state): State<AppState>) -> 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()
}