address mara: drop CompactionMode, always checkpoint-then-compact
This commit is contained in:
parent
fc42f97691
commit
c7bac398b2
4 changed files with 17 additions and 89 deletions
|
|
@ -40,7 +40,7 @@ somewhere."
|
||||||
| `Destroy` | For future use (`destroy --purge` does real I/O). Variant exists so the wire shape doesn't change later; not currently routed through the queue. |
|
| `Destroy` | For future use (`destroy --purge` does real I/O). Variant exists so the wire shape doesn't change later; not currently routed through the queue. |
|
||||||
| `Restart` | Stop + start a container without touching config (~5-10s). Routed through the queue so it serialises against in-flight rebuilds for the same agent — prevents a restart racing a rebuild mid-flight. Sources: dashboard ↺ button, the `restart` MCP tool. |
|
| `Restart` | Stop + start a container without touching config (~5-10s). Routed through the queue so it serialises against in-flight rebuilds for the same agent — prevents a restart racing a rebuild mid-flight. Sources: dashboard ↺ button, the `restart` MCP tool. |
|
||||||
| `PermChange` | Write a tool-group or capability change to the shared JSON file (`tool-groups.json` / `capabilities.json`), then rebuild the agent so the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes effect. Serialising the file write through the queue prevents concurrent dashboard batch-apply actions from racing on the shared file. After a successful file write, emits `CapabilitiesChanged` or `ToolGroupsChanged` SSE snapshot so the P3RM1SS10NS tab updates live. |
|
| `PermChange` | Write a tool-group or capability change to the shared JSON file (`tool-groups.json` / `capabilities.json`), then rebuild the agent so the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes effect. Serialising the file write through the queue prevents concurrent dashboard batch-apply actions from racing on the shared file. After a successful file write, emits `CapabilitiesChanged` or `ToolGroupsChanged` SSE snapshot so the P3RM1SS10NS tab updates live. |
|
||||||
| `GracefulStop` | Quiesce then stop a container (the `?graceful=true` path on `/api/kill/<agent>`). Signals the harness (its next `Recv` returns `GracefulStop` — the inbound fence — so it runs **one** stop-checkpoint turn that flushes durable `/state` and compacts the session if it crossed the watermark, then exits) and **immediately releases the build lane**, spawning a detached watcher that holds the `Stopping` transient across the drain (bounded by a 3-min timeout → hard-stop fallback) and then enqueues a fast-lane `Stop` (`parent_id` = this entry) for the actual `nixos-container stop`. Net: a whole-hive graceful stop signals every agent up front, drains overlap, and only the container teardowns serialise (on the fast lane). Queued so the signal can't race an in-flight rebuild for the same agent. |
|
| `GracefulStop` | Quiesce then stop a container (the `?graceful=true` path on `/api/kill/<agent>`). Signals the harness (its next `Recv` returns `GracefulStop` — the inbound fence — so it runs a stop-checkpoint turn that flushes durable `/state`, then takes the normal post-turn compaction path if it crossed the watermark, then exits) and **immediately releases the build lane**, spawning a detached watcher that holds the `Stopping` transient across the drain (bounded by a 3-min timeout → hard-stop fallback) and then enqueues a fast-lane `Stop` (`parent_id` = this entry) for the actual `nixos-container stop`. Net: a whole-hive graceful stop signals every agent up front, drains overlap, and only the container teardowns serialise (on the fast lane). Queued so the signal can't race an in-flight rebuild for the same agent. |
|
||||||
|
|
||||||
**Intentionally not queued** (sub-second ops): the *hard* `start`, `stop`, `kill`. (A *graceful* stop is the `GracefulStop` kind above — it takes a checkpoint turn, so it rides the queue.)
|
**Intentionally not queued** (sub-second ops): the *hard* `start`, `stop`, `kill`. (A *graceful* stop is the `GracefulStop` kind above — it takes a checkpoint turn, so it rides the queue.)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -517,12 +517,10 @@ ttl_seconds?, to?)`, `answer(id, answer)`.
|
||||||
explicit `from: "graceful-stop"` message instead of an empty inbox.
|
explicit `from: "graceful-stop"` message instead of an empty inbox.
|
||||||
This unmissably directs the agent to flush durable state (`/state`
|
This unmissably directs the agent to flush durable state (`/state`
|
||||||
files) and end the turn — the container exits when the turn completes.
|
files) and end the turn — the container exits when the turn completes.
|
||||||
This is a **single** wake: the graceful-stop turn *is* the checkpoint
|
The graceful-stop turn takes the same post-turn compaction path as any
|
||||||
(its prompt carries the full notes/CLAUDE.md/TODO.md flush guidance),
|
other turn: if the context crossed the watermark the harness runs a
|
||||||
so the harness compacts directly afterwards if the context crossed the
|
notes-checkpoint turn and then `/compact`. Compacting before shutdown
|
||||||
watermark (`CompactionMode::CompactOnly`) rather than waking a second
|
keeps a later cold start cheap instead of re-uploading a huge transcript.
|
||||||
dedicated checkpoint turn. Compacting before shutdown keeps a later
|
|
||||||
cold start cheap instead of re-uploading a huge transcript.
|
|
||||||
- `ask` — surface a structured question to the operator (default) or
|
- `ask` — surface a structured question to the operator (default) or
|
||||||
a peer agent (`to: "<agent>"`). Non-blocking — returns a question
|
a peer agent (`to: "<agent>"`). Non-blocking — returns a question
|
||||||
id; the answer arrives as a `question_answered` system event in the
|
id; the answer arrives as a `question_answered` system event in the
|
||||||
|
|
|
||||||
|
|
@ -546,9 +546,6 @@ async fn serve_loop<S: Surface>(
|
||||||
stats.as_ref(),
|
stats.as_ref(),
|
||||||
files,
|
files,
|
||||||
&turn_lock,
|
&turn_lock,
|
||||||
// The graceful prompt IS the checkpoint — compact
|
|
||||||
// directly afterwards if needed, no second wake.
|
|
||||||
turn::CompactionMode::CompactOnly,
|
|
||||||
graceful_stop_message(),
|
graceful_stop_message(),
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
|
|
@ -557,16 +554,7 @@ async fn serve_loop<S: Surface>(
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
let ctrl = handle_turn::<S>(
|
let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, next).await;
|
||||||
socket,
|
|
||||||
&bus,
|
|
||||||
stats.as_ref(),
|
|
||||||
files,
|
|
||||||
&turn_lock,
|
|
||||||
turn::CompactionMode::CheckpointThenCompact,
|
|
||||||
next,
|
|
||||||
)
|
|
||||||
.await;
|
|
||||||
if ctrl.auth_failed {
|
if ctrl.auth_failed {
|
||||||
*login_state.lock().unwrap() = LoginState::NeedsLogin;
|
*login_state.lock().unwrap() = LoginState::NeedsLogin;
|
||||||
turn::wait_for_login(
|
turn::wait_for_login(
|
||||||
|
|
@ -595,7 +583,6 @@ async fn handle_turn<S: Surface>(
|
||||||
stats: Option<&TurnStats>,
|
stats: Option<&TurnStats>,
|
||||||
files: &turn::TurnFiles,
|
files: &turn::TurnFiles,
|
||||||
turn_lock: &TurnLock,
|
turn_lock: &TurnLock,
|
||||||
compaction: turn::CompactionMode,
|
|
||||||
first: hive_sh4re::DeliveredMessage,
|
first: hive_sh4re::DeliveredMessage,
|
||||||
) -> TurnControl {
|
) -> TurnControl {
|
||||||
let from = first.from;
|
let from = first.from;
|
||||||
|
|
@ -616,7 +603,7 @@ async fn handle_turn<S: Surface>(
|
||||||
let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
|
let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
|
||||||
let outcome = {
|
let outcome = {
|
||||||
let _guard = turn_lock.lock().await;
|
let _guard = turn_lock.lock().await;
|
||||||
turn::drive_turn_with(&prompt, files, bus, compaction).await
|
turn::drive_turn(&prompt, files, bus).await
|
||||||
};
|
};
|
||||||
turn::emit_turn_end(bus, &outcome);
|
turn::emit_turn_end(bus, &outcome);
|
||||||
bus.set_state(TurnState::Idle);
|
bus.set_state(TurnState::Idle);
|
||||||
|
|
|
||||||
|
|
@ -263,20 +263,6 @@ fn compact_watermark_tokens(bus: &Bus) -> u64 {
|
||||||
effective_context_window(bus) * 3 / 4
|
effective_context_window(bus) * 3 / 4
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Post-turn compaction behaviour for [`drive_turn`].
|
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
||||||
pub enum CompactionMode {
|
|
||||||
/// Normal turns: if the context crossed the watermark, run a dedicated
|
|
||||||
/// notes-checkpoint turn so the agent can flush durable state, THEN
|
|
||||||
/// `/compact`. The checkpoint turn is a separate wake.
|
|
||||||
CheckpointThenCompact,
|
|
||||||
/// Graceful-stop turn: the turn that just ran WAS the checkpoint (its
|
|
||||||
/// prompt already told the agent to flush state), so compact directly
|
|
||||||
/// if the watermark was crossed — no second wake. Keeps a cold-start
|
|
||||||
/// resume cheap without waking the agent twice.
|
|
||||||
CompactOnly,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Drive one turn end-to-end. Three paths layer on top of the raw `run_turn`:
|
/// Drive one turn end-to-end. Three paths layer on top of the raw `run_turn`:
|
||||||
///
|
///
|
||||||
/// - **Auto-reset (pre-turn)** — context is large AND the prompt cache has
|
/// - **Auto-reset (pre-turn)** — context is large AND the prompt cache has
|
||||||
|
|
@ -292,24 +278,12 @@ pub enum CompactionMode {
|
||||||
/// size has crept past the watermark: while the session is still healthy we
|
/// size has crept past the watermark: while the session is still healthy we
|
||||||
/// give the agent one dedicated turn to checkpoint its `/state` notes, then
|
/// give the agent one dedicated turn to checkpoint its `/state` notes, then
|
||||||
/// compact. This keeps a later turn from hitting the reactive path (where
|
/// compact. This keeps a later turn from hitting the reactive path (where
|
||||||
/// there is no chance to save anything first). The graceful-stop path uses
|
/// there is no chance to save anything first). The graceful-stop path takes
|
||||||
/// [`CompactionMode::CompactOnly`] to skip the dedicated checkpoint turn.
|
/// the same proactive route — a checkpoint turn before `/compact` is cheap
|
||||||
|
/// insurance and keeps a later cold-start resume small.
|
||||||
///
|
///
|
||||||
/// Called once per turn by the `hive` serve loop (every agent role).
|
/// Called once per turn by the `hive` serve loop (every agent role).
|
||||||
pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
|
pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
|
||||||
drive_turn_with(prompt, files, bus, CompactionMode::CheckpointThenCompact).await
|
|
||||||
}
|
|
||||||
|
|
||||||
/// As [`drive_turn`] but with explicit control over the post-turn
|
|
||||||
/// compaction behaviour — see [`CompactionMode`]. The graceful-stop path
|
|
||||||
/// passes [`CompactionMode::CompactOnly`] so the agent isn't woken a
|
|
||||||
/// second time for a checkpoint it already did.
|
|
||||||
pub async fn drive_turn_with(
|
|
||||||
prompt: &str,
|
|
||||||
files: &TurnFiles,
|
|
||||||
bus: &Bus,
|
|
||||||
compaction: CompactionMode,
|
|
||||||
) -> TurnOutcome {
|
|
||||||
maybe_auto_reset(bus);
|
maybe_auto_reset(bus);
|
||||||
let outcome = match run_turn(prompt, files, bus).await {
|
let outcome = match run_turn(prompt, files, bus).await {
|
||||||
TurnOutcome::PromptTooLong => {
|
TurnOutcome::PromptTooLong => {
|
||||||
|
|
@ -340,22 +314,13 @@ pub async fn drive_turn_with(
|
||||||
other => other,
|
other => other,
|
||||||
};
|
};
|
||||||
// Proactive: a turn just completed on a still-healthy session. If its
|
// Proactive: a turn just completed on a still-healthy session. If its
|
||||||
// context crossed the watermark, compact before a later turn overflows
|
// context crossed the watermark, run a separate notes-checkpoint turn so
|
||||||
// into the reactive path. In the normal mode this first runs a separate
|
// the agent can flush durable state, then compact before a later turn
|
||||||
// notes-checkpoint turn so the agent can flush durable state; in
|
// overflows into the reactive path. Best-effort — never changes the
|
||||||
// `CompactOnly` mode (graceful stop) the turn that just ran was itself
|
// outcome of the turn that already succeeded, but records it as
|
||||||
// the checkpoint, so we compact directly without a second wake.
|
// `Compacted` so turn stats can distinguish it from a plain `Ok`.
|
||||||
// Best-effort — never changes the outcome of the turn that already
|
if matches!(outcome, TurnOutcome::Ok) && maybe_checkpoint_and_compact(files, bus).await {
|
||||||
// succeeded, but records it as `Compacted` so turn stats can distinguish
|
return TurnOutcome::Compacted;
|
||||||
// it from a plain `Ok`.
|
|
||||||
if matches!(outcome, TurnOutcome::Ok) {
|
|
||||||
let compacted = match compaction {
|
|
||||||
CompactionMode::CheckpointThenCompact => maybe_checkpoint_and_compact(files, bus).await,
|
|
||||||
CompactionMode::CompactOnly => maybe_compact(files, bus).await,
|
|
||||||
};
|
|
||||||
if compacted {
|
|
||||||
return TurnOutcome::Compacted;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
outcome
|
outcome
|
||||||
}
|
}
|
||||||
|
|
@ -399,28 +364,6 @@ async fn maybe_checkpoint_and_compact(files: &TurnFiles, bus: &Bus) -> bool {
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Compact-only proactive path: if the context crossed the watermark,
|
|
||||||
/// `/compact` directly — NO preceding checkpoint turn. Used by the
|
|
||||||
/// graceful-stop path, where the turn that just ran already flushed
|
|
||||||
/// durable state (its prompt said so), so a second checkpoint wake would
|
|
||||||
/// be redundant. Compacting before the container stops keeps a later
|
|
||||||
/// cold-start resume cheap rather than re-uploading a huge transcript.
|
|
||||||
/// Returns `true` if compaction was attempted.
|
|
||||||
async fn maybe_compact(files: &TurnFiles, bus: &Bus) -> bool {
|
|
||||||
let Some(used) = watermark_crossed(bus) else {
|
|
||||||
return false;
|
|
||||||
};
|
|
||||||
let watermark = compact_watermark_tokens(bus);
|
|
||||||
bus.emit(LiveEvent::Note {
|
|
||||||
text: format!(
|
|
||||||
"context at {used} tokens (watermark {watermark}) — compacting before \
|
|
||||||
graceful stop so a later cold start resumes cheap"
|
|
||||||
),
|
|
||||||
});
|
|
||||||
do_compact(files, bus).await;
|
|
||||||
true
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Returns `Some(used_tokens)` when proactive compaction is enabled AND the
|
/// Returns `Some(used_tokens)` when proactive compaction is enabled AND the
|
||||||
/// last inference's context size has reached the watermark; `None` when
|
/// last inference's context size has reached the watermark; `None` when
|
||||||
/// compaction is disabled (watermark 0), there's no usage reading yet, or
|
/// compaction is disabled (watermark 0), there's no usage reading yet, or
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue