hive-runtime: replace the session when an ACP compact command fails
A `compact` command that errors, or sends nothing, used to leave the session as full as before: `compacted` stayed false, so every following turn ran the checkpoint and another `/compact` again, each one waiting out the 600s turn idle window when the command was silent. - The `/compact` turn gets its own idle bound, COMPACT_IDLE (3 min, or the turn's idle window if shorter), through the turn's existing watchdog, so a silent command is cancelled (or killed) like any stalled turn. - When the command fails or hits that bound, compaction falls back to the no-command path: the session is archived and the next turn starts a new one. The checkpoint turn runs there only if it has not already run in this compaction. - Compaction returns early when attaching produced a new session (a failed `session/load` or a never-answered first prompt): there is nothing in it to compact. Refs #4391
This commit is contained in:
atlas
mara
parent
cec35bfbb1
commit
b8ec0f4f85
2 changed files with 85 additions and 12 deletions
|
|
@ -37,6 +37,11 @@ const SETTLE_MAX: Duration = Duration::from_secs(5);
|
|||
/// killed.
|
||||
const CANCEL_GRACE: Duration = Duration::from_secs(10);
|
||||
|
||||
/// How long a `compact` command may send nothing before it is cancelled and
|
||||
/// the session is replaced instead: room for one summarising model call over a
|
||||
/// full window, under a third of the default 600s turn idle window.
|
||||
const COMPACT_IDLE: Duration = Duration::from_mins(3);
|
||||
|
||||
/// The command an agent advertises to compact its session.
|
||||
const COMPACT_COMMAND: &str = "compact";
|
||||
|
||||
|
|
@ -166,9 +171,9 @@ enum Stop {
|
|||
///
|
||||
/// Compaction, proactive once `policy` says so after a turn or on
|
||||
/// [`Runtime::compact`], runs the agent's advertised `compact` command as a
|
||||
/// prompt on the same session. An agent advertising none gets the policy's
|
||||
/// checkpoint turn instead, and the session is archived, so the next turn
|
||||
/// starts a new one.
|
||||
/// prompt on the same session. An agent advertising none, or whose command
|
||||
/// fails, gets the policy's checkpoint turn instead (unless it already ran),
|
||||
/// and the session is archived, so the next turn starts a new one.
|
||||
pub struct AcpRuntime<P: CompactionPolicy> {
|
||||
command: AcpCommand,
|
||||
session_file: PathBuf,
|
||||
|
|
@ -177,6 +182,7 @@ pub struct AcpRuntime<P: CompactionPolicy> {
|
|||
live: Mutex<Option<Live>>,
|
||||
cancel: Arc<CancelState>,
|
||||
cancel_grace: Duration,
|
||||
compact_idle: Duration,
|
||||
}
|
||||
|
||||
/// The running agent process and the session loaded into it.
|
||||
|
|
@ -210,6 +216,7 @@ impl<P: CompactionPolicy> AcpRuntime<P> {
|
|||
live: Mutex::new(None),
|
||||
cancel: Arc::default(),
|
||||
cancel_grace: CANCEL_GRACE,
|
||||
compact_idle: COMPACT_IDLE,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -473,7 +480,9 @@ impl<P: CompactionPolicy> AcpRuntime<P> {
|
|||
/// Compact the recorded session, if there is one. `checkpoint` runs the
|
||||
/// policy's checkpoint turn before an advertised `compact` command too;
|
||||
/// without one it always runs, since archiving keeps nothing of the
|
||||
/// session.
|
||||
/// session. A `compact` command that fails, or is silent for
|
||||
/// `compact_idle`, is followed by the same archiving, so a compaction
|
||||
/// never leaves the session as full as it found it.
|
||||
async fn compact_session(
|
||||
&self,
|
||||
guard: &mut Option<Live>,
|
||||
|
|
@ -486,17 +495,37 @@ impl<P: CompactionPolicy> AcpRuntime<P> {
|
|||
}
|
||||
let live = self.running(guard, config).await?;
|
||||
let (servers, _) = mcp_servers(config)?;
|
||||
self.attach(live, &session_cwd(config), &servers).await?;
|
||||
let (_, created) = self.attach(live, &session_cwd(config), &servers).await?;
|
||||
if created {
|
||||
return Ok(());
|
||||
}
|
||||
let mut checkpointed = false;
|
||||
if advertises_compact(live).await? {
|
||||
if checkpoint {
|
||||
self.checkpoint(guard, config, sink).await;
|
||||
checkpointed = true;
|
||||
}
|
||||
self.prompt(guard, config, &format!("/{COMPACT_COMMAND}"), sink)
|
||||
.await?;
|
||||
return Ok(());
|
||||
let idle = config
|
||||
.idle_timeout
|
||||
.map_or(self.compact_idle, |idle| idle.min(self.compact_idle));
|
||||
let bounded = Config {
|
||||
idle_timeout: Some(idle),
|
||||
..config.clone()
|
||||
};
|
||||
let command = format!("/{COMPACT_COMMAND}");
|
||||
let Err(e) = self.prompt(guard, &bounded, &command, sink).await else {
|
||||
return Ok(());
|
||||
};
|
||||
tracing::warn!(error = %e, "ACP compact command failed; starting a new session");
|
||||
sink.on_stderr_line(&format!(
|
||||
"ACP {command} failed, starting a new session: {e}"
|
||||
));
|
||||
} else {
|
||||
tracing::info!("ACP agent advertises no compact command; starting a new session");
|
||||
}
|
||||
if !checkpointed {
|
||||
self.checkpoint(guard, config, sink).await;
|
||||
}
|
||||
tracing::info!("ACP agent advertises no compact command; starting a new session");
|
||||
self.checkpoint(guard, config, sink).await;
|
||||
self.archive()?;
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -720,7 +749,8 @@ mod tests {
|
|||
///
|
||||
/// With `COMMANDS` set, it advertises that one command on each session it
|
||||
/// creates or loads. With `USED` set, it reports that many of 1000 context
|
||||
/// tokens used on each prompt.
|
||||
/// tokens used on each prompt to session `s1`. With `STALL_COMPACT` set, it
|
||||
/// answers `/compact` only when cancelled, as `silent` does.
|
||||
const AGENT: &str = r#"
|
||||
n=0 prompt=
|
||||
printf 'start\n' >> "$1.methods"
|
||||
|
|
@ -747,7 +777,10 @@ while IFS= read -r line; do
|
|||
advertise "$sid" ;;
|
||||
session/prompt)
|
||||
printf '%s\n' "$line" >> "$1"
|
||||
[ -n "$USED" ] && update "$sid" "{\"sessionUpdate\":\"usage_update\",\"used\":$USED,\"size\":1000}"
|
||||
[ -n "$USED" ] && [ "$sid" = s1 ] && update "$sid" "{\"sessionUpdate\":\"usage_update\",\"used\":$USED,\"size\":1000}"
|
||||
case $line in
|
||||
*'"text":"/compact"'*) [ -n "$STALL_COMPACT" ] && prompt=$id && continue ;;
|
||||
esac
|
||||
if [ -e "$1.once" ]; then
|
||||
printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id"
|
||||
else
|
||||
|
|
@ -1101,4 +1134,40 @@ done
|
|||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_silent_compact_command_is_cut_short_and_a_new_session_started() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let env = [
|
||||
("COMMANDS", "compact"),
|
||||
("USED", "800"),
|
||||
("STALL_COMPACT", "1"),
|
||||
];
|
||||
let mut runtime = agent(dir.path(), "ok", &env, policy(75));
|
||||
runtime.compact_idle = Duration::from_millis(200);
|
||||
// The turn's own idle window is far longer than the compact bound.
|
||||
let config = idle_config(dir.path(), 5000);
|
||||
|
||||
let done = tokio::time::timeout(
|
||||
Duration::from_secs(3),
|
||||
runtime.run(&config, "one", &NoopSink),
|
||||
)
|
||||
.await
|
||||
.expect("the compact bound, not the turn's idle window, ends the /compact")
|
||||
.unwrap();
|
||||
assert!(done.compacted);
|
||||
let next = runtime.run(&config, "two", &NoopSink).await.unwrap();
|
||||
|
||||
assert!(next.created && !next.compacted);
|
||||
assert_eq!(
|
||||
prompts(dir.path()),
|
||||
[
|
||||
sent("s1", "SYSTEM PROMPT\n\none"),
|
||||
sent("s1", "CHECKPOINT"),
|
||||
sent("s1", "/compact"),
|
||||
sent("s2", "SYSTEM PROMPT\n\ntwo"),
|
||||
]
|
||||
);
|
||||
assert_eq!(recorded(dir.path()), "s2");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue