diff --git a/hive-agent-mcp/src/mcp/args.rs b/hive-agent-mcp/src/mcp/args.rs index 6a3f5442..1f828e76 100644 --- a/hive-agent-mcp/src/mcp/args.rs +++ b/hive-agent-mcp/src/mcp/args.rs @@ -80,6 +80,19 @@ pub struct RemindArgs { pub file_path: Option, } +/// MCP tool args for `compact`. +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +pub struct CompactArgs { + /// Optional wake-up prompt. When set and the compact actually runs + /// (gated on context usage — see the tool description), the harness + /// drives one synthetic follow-up turn with this string as its body + /// as soon as compaction finishes, so you don't have to wait for the + /// next external event to continue. Omit for a fire-and-forget compact + /// with no follow-up. + #[serde(default)] + pub wake_prompt: Option, +} + // ----------------------------------------------------------------------------- // Privileged tool arg types (lifecycle, approvals, scheduling, diagnostics) // ----------------------------------------------------------------------------- diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 4366e59a..75a86481 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -25,10 +25,10 @@ mod render; pub use args::{ AckUntilArgs, AgentGetLooseEndsArgs, AnswerArgs, AskArgs, CancelLooseEndArgs, - CancelScheduleArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, GetAgentMetaArgs, - GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, RemindArgs, - RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs, SetStatusArgs, - StartArgs, UpdateArgs, UpdateMetaInputsArgs, + CancelScheduleArgs, CompactArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, + GetAgentMetaArgs, GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, + RemindArgs, RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs, + SetStatusArgs, StartArgs, UpdateArgs, UpdateMetaInputsArgs, }; pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; @@ -657,11 +657,19 @@ impl AgentServer { that the call is refused with an explanation and has no effect. On a pass, \ queues compaction for the end of the current turn (same deferred mechanism \ the dashboard button uses, so it never races a live claude process); the \ - usual pre-compaction notes-checkpoint turn still fires first. No args." + usual pre-compaction notes-checkpoint turn still fires first. Pass \ + `wake_prompt` to have the harness drive one synthetic follow-up turn with \ + that body as soon as compaction finishes — without it you just go idle \ + waiting for the next external event, same as ending a turn normally." )] - async fn compact(&self) -> String { - run_tool_envelope("compact", String::new(), async move { - match dial_agent_socket(&hive_agent_sock::Request::Compact).await { + async fn compact(&self, Parameters(args): Parameters) -> String { + let log = format!("{args:?}"); + run_tool_envelope("compact", log, async move { + match dial_agent_socket(&hive_agent_sock::Request::Compact { + wake_prompt: args.wake_prompt, + }) + .await + { Some(hive_agent_sock::Response::Ok) => { "compact queued — will run at the end of the current turn".to_owned() } diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index af436056..8fb6641b 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -117,7 +117,12 @@ pub enum Request { /// explains why and takes no action. On a pass, queues the same /// deferred `compact_pending` flag the operator's button sets (consumed /// at the next turn boundary), so it never races a live claude process. - Compact, + /// `wake_prompt`, when set, is driven as a synthetic follow-up turn once + /// the compaction actually finishes — the dashboard button's own + /// requests go through `Bus::request_compact` directly with `None`, not + /// through this variant, since a human watching the dashboard isn't + /// waiting on a wake. + Compact { wake_prompt: Option }, /// Mirror an outstanding question this agent asked (`ask()` succeeded). /// `target` is who it's waiting on (`"operator"` when asked with /// `to: None`). Part of the questions-mirror increment — see diff --git a/hive-agent/src/events.rs b/hive-agent/src/events.rs index cf585f5f..e9a26990 100644 --- a/hive-agent/src/events.rs +++ b/hive-agent/src/events.rs @@ -260,6 +260,14 @@ pub enum TurnState { Compacting, } +/// One pending `/compact` request (see `Bus::request_compact`). +/// `wake_prompt` is what to drive as a synthetic follow-up turn once the +/// compaction actually finishes, if anything. +#[derive(Debug, Clone)] +pub struct CompactRequest { + pub wake_prompt: Option, +} + #[derive(Clone)] pub struct Bus { tx: Arc>, @@ -307,7 +315,23 @@ pub struct Bus { /// 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, + /// `Some(request)` when a compact is pending; `request.wake_prompt` is + /// what to drive as a synthetic follow-up turn once the compaction + /// actually completes (`None` = pending but no follow-up wake wanted, + /// e.g. the operator dashboard's `/compact` button). `None` = no compact + /// pending. Wrapped in [`CompactRequest`] rather than + /// `Option>` (clippy pedantic's `option_option` lint, + /// and the named field reads clearer at call sites than a bare nested + /// `Option`) so "pending" and "what to wake with" can never desync. + compact_pending: Arc>>, + /// One-shot, written by `turn::drive_turn`/`turn::run_pending_compact` + /// right after a compaction they served finishes, when that compact's + /// request carried a `wake_prompt`. Read once by the `hive-agent` serve + /// loop after either call site to decide whether to drive a synthetic + /// follow-up turn. Separate from `compact_pending`: by the time this is + /// set, the compact has already run and that flag has already been + /// cleared by `take_compact`. + post_compact_wake: 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 @@ -397,7 +421,8 @@ 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)), + compact_pending: Arc::new(Mutex::new(None)), + post_compact_wake: Arc::new(Mutex::new(None)), session_id: Arc::new(Mutex::new(None)), fresh_session: Arc::new(AtomicBool::new(false)), tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())), @@ -438,16 +463,40 @@ impl Bus { } /// 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); + /// boundary). Idempotent — a second request before the first is + /// serviced just overwrites `wake_prompt` with the latest ask. `Some + /// (wake_prompt)` schedules a synthetic follow-up turn (driven with + /// `wake_prompt` as its body) once the compaction actually completes; + /// `None` requests a plain compact with no follow-up wake (the operator + /// dashboard's `/compact` button). + pub fn request_compact(&self, wake_prompt: Option) { + *self.compact_pending.lock().unwrap() = Some(CompactRequest { wake_prompt }); } - /// Take + clear the compact one-shot. Returns true iff `drive_turn` should - /// compact at the end of this turn. + /// Take + clear the compact one-shot. `Some(request)` means + /// `drive_turn`/`run_pending_compact` should compact now — + /// `request.wake_prompt` is what to pass to `set_post_compact_wake` once + /// that compaction finishes. `None` means no compact is pending. #[must_use] - pub fn take_compact(&self) -> bool { - self.compact_pending.swap(false, Ordering::SeqCst) + pub fn take_compact(&self) -> Option { + self.compact_pending.lock().unwrap().take() + } + + /// Record that a just-finished compaction should drive a synthetic + /// follow-up turn with `prompt` as its body. Called by + /// `turn::drive_turn`/`turn::run_pending_compact` right after the + /// compaction they served (whose `take_compact()` returned a request + /// with `wake_prompt: Some(prompt)`) completes. + pub fn set_post_compact_wake(&self, prompt: String) { + *self.post_compact_wake.lock().unwrap() = Some(prompt); + } + + /// Take + clear the post-compact wake one-shot. The serve loop calls + /// this after either compact call site to decide whether to + /// synthesize a follow-up turn. + #[must_use] + pub fn take_post_compact_wake(&self) -> Option { + self.post_compact_wake.lock().unwrap().take() } /// Mark that the current turn started a fresh claude session. diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index ce888170..95106ff4 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -240,6 +240,21 @@ fn synthetic_todo_message(stern: bool) -> hive_sh4re::inbox::DeliveredMessage { } } +/// Synthetic message that drives the follow-up turn when a `/compact` call +/// (self-requested via the `compact` MCP tool's `wake_prompt` arg) finishes +/// and asked to be woken. Mirrors `synthetic_todo_message`'s "no broker row" +/// sentinel shape (`id = 0`) — the compact tool call itself is the durable +/// record that a wake was requested, not a broker message. +fn post_compact_wake_message(prompt: String) -> hive_sh4re::inbox::DeliveredMessage { + hive_sh4re::inbox::DeliveredMessage { + from: "compact".into(), + body: prompt, + id: 0, + redelivered: false, + in_reply_to: None, + } +} + /// Synthetic message that drives the single stop-checkpoint turn when c0re /// signals a graceful stop. The agent gets one final turn to flush durable /// `/state` before the container is stopped; new inbound is already fenced. @@ -772,8 +787,18 @@ async fn serve_loop( let compacted = turn::run_pending_compact(files, &bus, &session).await; if !compacted { tokio::time::sleep(interval).await; + continue; } - continue; + // The compact that just ran may have carried a wake prompt + // (the agent's own `compact` tool, not the operator button — + // see `Bus::request_compact`). If so, drive it as a synthetic + // turn right now instead of looping back to `recv_next` and + // waiting for the next external event. + let Some(prompt) = bus.take_post_compact_wake() else { + continue; + }; + tracing::debug!("post-compact wake queued, driving synthetic follow-up turn"); + post_compact_wake_message(prompt) } RecvOutcome::TransportError => { // `recv_next` already logged the detail; just retry. @@ -802,32 +827,83 @@ async fn serve_loop( return Ok(()); } }; - let ctrl = handle_turn::( + let turn_ctx = TurnCtx { socket, - &bus, - stats.as_ref(), + bus: &bus, + stats: stats.as_ref(), files, - &session, + session: &session, + interrupted: &interrupted, + login_state: &login_state, + claude_dir: &claude_dir, + interval, + }; + drive_turn_and_wake_chain::(&turn_ctx, next, &mut todo_miss_streak).await; + } +} + +/// Loop-invariant turn-driving context for `drive_turn_and_wake_chain`, +/// threaded as one bundle instead of double-digit positional args +/// (clippy's `too_many_arguments`). Everything here is constant for the +/// lifetime of one `serve_loop` call; only the message to drive and the +/// todo-miss streak vary per turn and stay as separate params. +struct TurnCtx<'a> { + socket: &'a Path, + bus: &'a Bus, + stats: Option<&'a TurnStats>, + files: &'a turn::TurnFiles, + session: &'a turn::AgentSession, + interrupted: &'a Arc, + login_state: &'a Arc>, + claude_dir: &'a Path, + interval: Duration, +} + +/// Drive `next`, then keep driving synthetic follow-up turns for as long as +/// a compact that just ran carries a wake prompt +/// (`Bus::take_post_compact_wake`) — see `serve_loop`'s comment at the call +/// site for why this loops in place instead of returning to the outer +/// `select!`/`recv_next`. Ordinarily runs exactly one iteration; only chains +/// further if the follow-up turn itself requests another woken compact. +/// Split out of `serve_loop` purely to keep that function under clippy's +/// line limit. +async fn drive_turn_and_wake_chain( + ctx: &TurnCtx<'_>, + mut next: hive_sh4re::inbox::DeliveredMessage, + todo_miss_streak: &mut u32, +) { + loop { + let ctrl = handle_turn::( + ctx.socket, + ctx.bus, + ctx.stats, + ctx.files, + ctx.session, next, - &interrupted, + ctx.interrupted, ) .await; - apply_todo_wake_checked(ctrl.todo_wake_checked, &mut todo_miss_streak, &bus); + apply_todo_wake_checked(ctrl.todo_wake_checked, todo_miss_streak, ctx.bus); if ctrl.auth_failed { - *login_state.lock().unwrap() = LoginState::NeedsLogin; + *ctx.login_state.lock().unwrap() = LoginState::NeedsLogin; // Baseline the resume check on *this instant*, not on a // directory snapshot taken after `wait_for_login` starts // polling — closes the race where a login lands between the // 401 and the first poll. See `wait_for_login`'s doc comment. login::wait_for_login( - &claude_dir, - login_state.clone(), - &bus, - u64::try_from(interval.as_millis()).unwrap_or(2000), + ctx.claude_dir, + ctx.login_state.clone(), + ctx.bus, + u64::try_from(ctx.interval.as_millis()).unwrap_or(2000), std::time::SystemTime::now(), ) .await; } + let Some(prompt) = ctx.bus.take_post_compact_wake() else { + break; + }; + tracing::debug!("post-compact wake queued mid-turn-flow, driving synthetic follow-up turn"); + next = post_compact_wake_message(prompt); } } diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 0352787c..307d874d 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -242,7 +242,7 @@ fn dispatch( } => record_answering_question(questions, id, &asker, &question), Request::ClearQuestion { id } => clear_question(questions, id), Request::ListQuestions => list_questions(questions), - Request::Compact => compact(bus), + Request::Compact { wake_prompt } => compact(bus, wake_prompt), } } @@ -458,8 +458,10 @@ fn mark_todos_done(store: &Todos, ids: &[i64]) -> Response { /// the agent's own MCP tool instead of the dashboard, and refuses below /// [`COMPACT_MIN_USAGE_FRACTION`] instead of always honouring the request — /// an agent can call this speculatively, a human clicking the dashboard -/// button already made the judgment call. -fn compact(bus: &Bus) -> Response { +/// button already made the judgment call. `wake_prompt`, when set, is +/// forwarded to `Bus::request_compact` so the turn loop drives a synthetic +/// follow-up turn once the compaction actually finishes. +fn compact(bus: &Bus, wake_prompt: Option) -> Response { let Some(usage) = bus.last_ctx_usage() else { return Response::Err { message: "compact refused: no completed turn yet — nothing to compact".to_owned(), @@ -487,9 +489,16 @@ fn compact(bus: &Bus) -> Response { ), }; } - bus.request_compact(); + let will_wake = wake_prompt.is_some(); + bus.request_compact(wake_prompt); bus.emit(crate::events::LiveEvent::Note { - text: "agent: self-requested /compact — running at the end of the current turn".into(), + text: if will_wake { + "agent: self-requested /compact (with wake prompt) — running at the end of the \ + current turn" + .into() + } else { + "agent: self-requested /compact — running at the end of the current turn".into() + }, }); Response::Ok } diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index 226d1a13..8b8f4a23 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -374,15 +374,18 @@ pub async fn drive_turn( archive_session(bus); return Err(TurnError::PromptTooLong); } - // 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. - // `is_ok()` first: `take_compact()` clears the flag, so it must only fire - // when the compaction will actually run. On an unhealthy turn - // (rate-limited / auth-failed / failed) the flag is left set for the next - // turn or the idle `run_pending_compact` to service — not silently eaten. - if outcome.is_ok() && bus.take_compact() { + // Operator `/compact` (`POST /api/compact`) or an agent's own `compact` + // MCP tool call, 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. `is_ok()` first: `take_compact()` + // clears the flag, so it must only fire when the compaction will actually + // run. On an unhealthy turn (rate-limited / auth-failed / failed) the flag + // is left set for the next turn or the idle `run_pending_compact` to + // service — not silently eaten. + if outcome.is_ok() + && let Some(request) = bus.take_compact() + { bus.emit(LiveEvent::Note { text: "operator: /compact — running at turn end".into(), }); @@ -390,6 +393,14 @@ pub async fn drive_turn( // does; the serve loop resets to `Idle` once this turn returns. bus.set_state(crate::events::TurnState::Compacting); let _ = session.compact(&config, &sink).await; + // If the compact call asked to be woken (the agent's own `compact` + // tool with a `wake_prompt`), stash it — the serve loop reads it + // back after this turn returns and drives a synthetic follow-up + // turn, so a self-requested compact provably doesn't strand the + // agent idle waiting for the next external event. + if let Some(prompt) = request.wake_prompt { + bus.set_post_compact_wake(prompt); + } return Ok(true); } outcome @@ -500,11 +511,13 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { /// 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. +/// `true` if a compaction ran; the serve loop follows up with +/// `Bus::take_post_compact_wake` to see whether a synthetic follow-up turn +/// should run (set below when the compact request carried a `wake_prompt`). pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSession) -> bool { - if !bus.take_compact() { + let Some(request) = bus.take_compact() else { return false; - } + }; bus.emit(LiveEvent::Note { text: "operator: /compact — running on idle session".into(), }); @@ -520,6 +533,9 @@ pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSe }), } bus.set_state(crate::events::TurnState::Idle); + if let Some(prompt) = request.wake_prompt { + bus.set_post_compact_wake(prompt); + } true } diff --git a/hive-agent/src/web_ui/actions.rs b/hive-agent/src/web_ui/actions.rs index f05905e0..c5e7d115 100644 --- a/hive-agent/src/web_ui/actions.rs +++ b/hive-agent/src/web_ui/actions.rs @@ -80,7 +80,10 @@ pub(super) async fn post_cancel_turn(State(state): State) -> Response /// claude process) rather than only when the agent is idle. Returns 200 /// immediately; the compaction stream lands in the live panel when it runs. pub(super) async fn post_compact(State(state): State) -> Response { - state.bus.request_compact(); + // No wake prompt: the operator is watching the dashboard, not waiting on + // an inbox message — `request_compact`'s wake-prompt arg exists for the + // agent's own `compact` MCP tool (`todo_server.rs::compact`). + state.bus.request_compact(None); state.bus.emit(crate::events::LiveEvent::Note { text: "operator: /compact queued — runs at the end of the current turn".into(), });