hive-agent: guarantee a wake after a self-requested /compact

This commit is contained in:
damocles 2026-08-13 21:30:26 +02:00 committed by mara
commit 2c7872841a
8 changed files with 227 additions and 48 deletions

View file

@ -80,6 +80,19 @@ pub struct RemindArgs {
pub file_path: Option<String>, pub file_path: Option<String>,
} }
/// 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<String>,
}
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
// Privileged tool arg types (lifecycle, approvals, scheduling, diagnostics) // Privileged tool arg types (lifecycle, approvals, scheduling, diagnostics)
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------

View file

@ -25,10 +25,10 @@ mod render;
pub use args::{ pub use args::{
AckUntilArgs, AgentGetLooseEndsArgs, AnswerArgs, AskArgs, CancelLooseEndArgs, AckUntilArgs, AgentGetLooseEndsArgs, AnswerArgs, AskArgs, CancelLooseEndArgs,
CancelScheduleArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, GetAgentMetaArgs, CancelScheduleArgs, CompactArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs,
GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, RemindArgs, GetAgentMetaArgs, GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs,
RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs, SetStatusArgs, RemindArgs, RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs,
StartArgs, UpdateArgs, UpdateMetaInputsArgs, SetStatusArgs, StartArgs, UpdateArgs, UpdateMetaInputsArgs,
}; };
pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; 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, \ 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 \ 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 \ 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 { async fn compact(&self, Parameters(args): Parameters<CompactArgs>) -> String {
run_tool_envelope("compact", String::new(), async move { let log = format!("{args:?}");
match dial_agent_socket(&hive_agent_sock::Request::Compact).await { 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) => { Some(hive_agent_sock::Response::Ok) => {
"compact queued — will run at the end of the current turn".to_owned() "compact queued — will run at the end of the current turn".to_owned()
} }

View file

@ -117,7 +117,12 @@ pub enum Request {
/// explains why and takes no action. On a pass, queues the same /// explains why and takes no action. On a pass, queues the same
/// deferred `compact_pending` flag the operator's button sets (consumed /// deferred `compact_pending` flag the operator's button sets (consumed
/// at the next turn boundary), so it never races a live claude process. /// 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<String> },
/// Mirror an outstanding question this agent asked (`ask()` succeeded). /// Mirror an outstanding question this agent asked (`ask()` succeeded).
/// `target` is who it's waiting on (`"operator"` when asked with /// `target` is who it's waiting on (`"operator"` when asked with
/// `to: None`). Part of the questions-mirror increment — see /// `to: None`). Part of the questions-mirror increment — see

View file

@ -260,6 +260,14 @@ pub enum TurnState {
Compacting, 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<String>,
}
#[derive(Clone)] #[derive(Clone)]
pub struct Bus { pub struct Bus {
tx: Arc<broadcast::Sender<BusEvent>>, tx: Arc<broadcast::Sender<BusEvent>>,
@ -307,7 +315,23 @@ pub struct Bus {
/// One-shot: run `/compact` after the next turn ends. Consumed at the end /// 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 /// of the current/next turn by `turn::drive_turn`. Deferring to the turn
/// boundary keeps compaction from racing a live claude process mid-turn. /// boundary keeps compaction from racing a live claude process mid-turn.
compact_pending: Arc<AtomicBool>, /// `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<Option<String>>` (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<Mutex<Option<CompactRequest>>>,
/// 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<Mutex<Option<String>>>,
/// Current fresh-claude-session id (FK to `sessions.id`). Set by the /// 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 /// bin loop after minting a session row on a fresh start; stamped onto
/// every `turn_stats` row until the next fresh session. `None` before /// 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)), last_cost_usage: Arc::new(Mutex::new(None)),
rate_limited: Arc::new(AtomicBool::new(was_rate_limited)), rate_limited: Arc::new(AtomicBool::new(was_rate_limited)),
session_reset_pending: Arc::new(AtomicBool::new(false)), 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)), session_id: Arc::new(Mutex::new(None)),
fresh_session: Arc::new(AtomicBool::new(false)), fresh_session: Arc::new(AtomicBool::new(false)),
tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())), 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 /// Request a compaction after the next turn ends (deferred to the turn
/// boundary). Idempotent. /// boundary). Idempotent — a second request before the first is
pub fn request_compact(&self) { /// serviced just overwrites `wake_prompt` with the latest ask. `Some
self.compact_pending.store(true, Ordering::SeqCst); /// (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<String>) {
*self.compact_pending.lock().unwrap() = Some(CompactRequest { wake_prompt });
} }
/// Take + clear the compact one-shot. Returns true iff `drive_turn` should /// Take + clear the compact one-shot. `Some(request)` means
/// compact at the end of this turn. /// `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] #[must_use]
pub fn take_compact(&self) -> bool { pub fn take_compact(&self) -> Option<CompactRequest> {
self.compact_pending.swap(false, Ordering::SeqCst) 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<String> {
self.post_compact_wake.lock().unwrap().take()
} }
/// Mark that the current turn started a fresh claude session. /// Mark that the current turn started a fresh claude session.

View file

@ -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 /// Synthetic message that drives the single stop-checkpoint turn when c0re
/// signals a graceful stop. The agent gets one final turn to flush durable /// signals a graceful stop. The agent gets one final turn to flush durable
/// `/state` before the container is stopped; new inbound is already fenced. /// `/state` before the container is stopped; new inbound is already fenced.
@ -772,8 +787,18 @@ async fn serve_loop<S: Surface>(
let compacted = turn::run_pending_compact(files, &bus, &session).await; let compacted = turn::run_pending_compact(files, &bus, &session).await;
if !compacted { if !compacted {
tokio::time::sleep(interval).await; 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 => { RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry. // `recv_next` already logged the detail; just retry.
@ -802,32 +827,83 @@ async fn serve_loop<S: Surface>(
return Ok(()); return Ok(());
} }
}; };
let ctrl = handle_turn::<S>( let turn_ctx = TurnCtx {
socket, socket,
&bus, bus: &bus,
stats.as_ref(), stats: stats.as_ref(),
files, files,
&session, session: &session,
interrupted: &interrupted,
login_state: &login_state,
claude_dir: &claude_dir,
interval,
};
drive_turn_and_wake_chain::<S>(&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<std::sync::atomic::AtomicBool>,
login_state: &'a Arc<Mutex<LoginState>>,
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<S: Surface>(
ctx: &TurnCtx<'_>,
mut next: hive_sh4re::inbox::DeliveredMessage,
todo_miss_streak: &mut u32,
) {
loop {
let ctrl = handle_turn::<S>(
ctx.socket,
ctx.bus,
ctx.stats,
ctx.files,
ctx.session,
next, next,
&interrupted, ctx.interrupted,
) )
.await; .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 { 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 // Baseline the resume check on *this instant*, not on a
// directory snapshot taken after `wait_for_login` starts // directory snapshot taken after `wait_for_login` starts
// polling — closes the race where a login lands between the // polling — closes the race where a login lands between the
// 401 and the first poll. See `wait_for_login`'s doc comment. // 401 and the first poll. See `wait_for_login`'s doc comment.
login::wait_for_login( login::wait_for_login(
&claude_dir, ctx.claude_dir,
login_state.clone(), ctx.login_state.clone(),
&bus, ctx.bus,
u64::try_from(interval.as_millis()).unwrap_or(2000), u64::try_from(ctx.interval.as_millis()).unwrap_or(2000),
std::time::SystemTime::now(), std::time::SystemTime::now(),
) )
.await; .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);
} }
} }

View file

@ -242,7 +242,7 @@ fn dispatch(
} => record_answering_question(questions, id, &asker, &question), } => record_answering_question(questions, id, &asker, &question),
Request::ClearQuestion { id } => clear_question(questions, id), Request::ClearQuestion { id } => clear_question(questions, id),
Request::ListQuestions => list_questions(questions), 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 /// the agent's own MCP tool instead of the dashboard, and refuses below
/// [`COMPACT_MIN_USAGE_FRACTION`] instead of always honouring the request — /// [`COMPACT_MIN_USAGE_FRACTION`] instead of always honouring the request —
/// an agent can call this speculatively, a human clicking the dashboard /// an agent can call this speculatively, a human clicking the dashboard
/// button already made the judgment call. /// button already made the judgment call. `wake_prompt`, when set, is
fn compact(bus: &Bus) -> Response { /// 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<String>) -> Response {
let Some(usage) = bus.last_ctx_usage() else { let Some(usage) = bus.last_ctx_usage() else {
return Response::Err { return Response::Err {
message: "compact refused: no completed turn yet — nothing to compact".to_owned(), 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 { 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 Response::Ok
} }

View file

@ -374,15 +374,18 @@ pub async fn drive_turn(
archive_session(bus); archive_session(bus);
return Err(TurnError::PromptTooLong); return Err(TurnError::PromptTooLong);
} }
// Operator `/compact` (`POST /api/compact`) deferred to the turn boundary: // Operator `/compact` (`POST /api/compact`) or an agent's own `compact`
// run it now that the turn is done, so it works mid-turn rather than only // MCP tool call, deferred to the turn boundary: run it now that the turn
// when the agent is idle. Only on a healthy turn — no point spawning a // is done, so it works mid-turn rather than only when the agent is idle.
// compaction after a rate-limited / auth-failed / crashed one. // Only on a healthy turn — no point spawning a compaction after a
// `is_ok()` first: `take_compact()` clears the flag, so it must only fire // rate-limited / auth-failed / crashed one. `is_ok()` first: `take_compact()`
// when the compaction will actually run. On an unhealthy turn // clears the flag, so it must only fire when the compaction will actually
// (rate-limited / auth-failed / failed) the flag is left set for the next // run. On an unhealthy turn (rate-limited / auth-failed / failed) the flag
// turn or the idle `run_pending_compact` to service — not silently eaten. // is left set for the next turn or the idle `run_pending_compact` to
if outcome.is_ok() && bus.take_compact() { // service — not silently eaten.
if outcome.is_ok()
&& let Some(request) = bus.take_compact()
{
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note {
text: "operator: /compact — running at turn end".into(), 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. // does; the serve loop resets to `Idle` once this turn returns.
bus.set_state(crate::events::TurnState::Compacting); bus.set_state(crate::events::TurnState::Compacting);
let _ = session.compact(&config, &sink).await; 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); return Ok(true);
} }
outcome 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 /// 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 /// case is handled at the end of [`drive_turn`].) Resume-only via
/// [`InfiniteSession::compact`]: a missing session is a harmless no-op. Returns /// [`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 { 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; return false;
} };
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note {
text: "operator: /compact — running on idle session".into(), 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); bus.set_state(crate::events::TurnState::Idle);
if let Some(prompt) = request.wake_prompt {
bus.set_post_compact_wake(prompt);
}
true true
} }

View file

@ -80,7 +80,10 @@ pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response
/// claude process) rather than only when the agent is idle. Returns 200 /// claude process) rather than only when the agent is idle. Returns 200
/// immediately; the compaction stream lands in the live panel when it runs. /// immediately; the compaction stream lands in the live panel when it runs.
pub(super) async fn post_compact(State(state): State<AppState>) -> Response { pub(super) async fn post_compact(State(state): State<AppState>) -> 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 { state.bus.emit(crate::events::LiveEvent::Note {
text: "operator: /compact queued — runs at the end of the current turn".into(), text: "operator: /compact queued — runs at the end of the current turn".into(),
}); });