Skip to content

Background-job survival, sync circuit breaker, sub-50ms startup #132

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davidrhodus merged 2 commits into from
Jul 26, 2026
Merged

Background-job survival, sync circuit breaker, sub-50ms startup #132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6ae18a4
fix: deliberate background jobs survive turn end, cancel, and rewinds
Jul 26, 2026
888e8ed
fix: a dead sync portal is now invisible — breaker, async host enable…
Jul 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion crates/hi-agent/src/agent/lifecycle.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -817,7 +817,11 @@ impl crate::Agent {
self.runtime.background().snapshot()
}

/// Kill only background processes this agent started after `before`.
/// Kill only **auto-backgrounded** processes (foreground commands that
/// outgrew their timeout) started after `before`. Deliberate
/// `run_in_background` jobs are spared — they are long-lived work that
/// must survive turn end, cancel, and retry rewinds; they still die with
/// the session or an explicit `bash_kill`.
pub fn kill_background_processes_started_after(&self, before: &[String]) -> usize {
self.runtime.background().kill_started_after(before)
}
Expand Down
5 changes: 4 additions & 1 deletion crates/hi-agent/src/agent/turn/verify_run.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,13 +26,16 @@ impl crate::Agent {
) -> Result<VerifyOutcome> {
// Caller stamps WorkspaceRepair before invoking; keep phase sticky here.
debug_assert_eq!(self.turn_phase(), TurnPhase::WorkspaceRepair);
// Reaps only auto-backgrounded foreground overruns. Deliberate
// `run_in_background` jobs (downloads, servers) survive the turn —
// killing them here once cost two ~800 GB downloads at turn end.
let killed_backgrounds = self
.runtime
.background()
.kill_started_after(turn_background_baseline);
if killed_backgrounds > 0 {
ui.status(&format!(
"stopped {killed_backgrounds} live background process(es) before final verification"
"stopped {killed_backgrounds} auto-backgrounded process(es) before final verification"
));
// Process-group termination is signalled synchronously. Yield so
// the driver tasks can observe it before the final filesystem
Expand Down
54 changes: 54 additions & 0 deletions crates/hi-agent/src/tests/turn.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1888,6 +1888,60 @@ async fn waiting_on_live_background_with_fresh_output_ends_with_status_report()
agent.messages.validate_for_provider().unwrap();
}

#[tokio::test]
async fn deliberate_background_process_survives_turn_end() {
// The sol-view regression: two ~800 GB downloads spawned with
// run_in_background were reaped by pre-verification turn-end cleanup
// hours before completion. Work the model deliberately backgrounds must
// outlive the turn that started it; only auto-backgrounded foreground
// overruns are turn state.
let responses = vec![
completion(
vec![Content::ToolCall {
id: "spawn".into(),
name: "bash".into(),
arguments: serde_json::json!({
"command": "sleep 600",
"run_in_background": true,
})
.to_string(),
}],
1,
1,
),
completion(
vec![Content::Text(
"The download is running in the background and will continue after this turn.".into(),
)],
1,
1,
),
];
let mut agent = agent(responses, config());
let mut ui = RecUi::default();

agent
.run_turn("start the big download in the background", &mut ui)
.await
.unwrap();

let ids = agent.runtime.background().ids();
assert_eq!(ids.len(), 1, "the spawned job is registered: {ids:?}");
assert_eq!(
agent.runtime.background().outcome(&ids[0]).unwrap().state,
hi_tools::BackgroundState::Running,
"a deliberate run_in_background job survives turn end"
);
assert!(
!ui.statuses
.iter()
.any(|s| s.contains("stopped") && s.contains("background")),
"turn end must not report reaping deliberate jobs: {:?}",
ui.statuses
);
let _ = agent.runtime.background().kill(&ids[0]);
}

#[tokio::test]
async fn repeated_completed_background_output_poll_is_bounded() {
let id = "bg_1".to_string();
Expand Down
77 changes: 53 additions & 24 deletions crates/hi-cli/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,19 @@ async fn run() -> Result<()> {
});
}

// Startup phase tracing: `HI_STARTUP_TRACE=1 hi …` prints elapsed-at-
// milestone lines to stderr, so slow-start regressions get measured
// instead of guessed at. Zero cost when the variable is unset.
let startup_began = std::time::Instant::now();
let startup_trace_on = std::env::var_os("HI_STARTUP_TRACE").is_some();
macro_rules! startup_trace {
($label:expr) => {
if startup_trace_on {
eprintln!("[startup {:>9.2?}] {}", startup_began.elapsed(), $label);
}
};
}

let raw_args = std::env::args().collect::<Vec<_>>();
if raw_args.get(1).map(String::as_str) == Some("hf") {
return run_hf_cli(&raw_args[2..]).await;
Expand Down Expand Up @@ -148,6 +161,7 @@ async fn run() -> Result<()> {
}

let cli = bootstrap::parse_and_validate_cli();
startup_trace!("cli parsed");
if cli.benchmark_orchestration {
orchestration_benchmark::run();
return Ok(());
Expand Down Expand Up @@ -237,7 +251,9 @@ async fn run() -> Result<()> {
}
let (workspace_root, state_root) =
resolve_runtime_roots().inspect_err(|error| report_init_failure(error, None))?;
startup_trace!("runtime roots resolved");
let recovered = scheduler_ops::recover_stale_state(&state_root);
startup_trace!("stale scheduler state recovered");
if recovered > 0 {
eprintln!("recovered {recovered} stale scheduler artifact(s)");
}
Expand All @@ -257,8 +273,10 @@ async fn run() -> Result<()> {
&std::collections::BTreeSet::new(),
)
.ok();
startup_trace!("background ledger scan launched");
let rsi = RsiBootstrap::initialize(&cli, &file, prompt_input.as_deref())
.inspect_err(|error| report_init_failure(error, None))?;
startup_trace!("rsi bootstrap initialized");
let rsi_requested = rsi.requested;
let quality = match config::resolve_quality(&cli, &workspace_root) {
Ok(quality) => quality,
Expand All @@ -267,7 +285,9 @@ async fn run() -> Result<()> {
std::process::exit(2);
}
};
startup_trace!("quality resolved");
let verify_stages = quality.verification.resolved_stages(&workspace_root);
startup_trace!("verify stages resolved");
if matches!(quality.verification, VerificationMode::Auto) && !verify_stages.is_empty() {
eprintln!(
"\x1b[2mverification: auto ({})\x1b[0m",
Expand Down Expand Up @@ -306,11 +326,13 @@ async fn run() -> Result<()> {
// Resolve which session file to use and any history to resume.
let (session_path, loaded) =
resolve_session(&cli).inspect_err(|error| report_init_failure(error, None))?;
startup_trace!("session resolved");
let mut feedback_session_id = feedback::session_id_from_path(&session_path);

let fallbacks = config::resolve_fallbacks(&cli, &file);
// Arc so the agent can share it with read-only `explore` subagents.
let base_provider: std::sync::Arc<dyn Provider> = build_chain(&settings, fallbacks).into();
startup_trace!("provider chain built");
let rsi_bundle = rsi_bootstrap::wrap_provider(
&cli,
&file,
Expand All @@ -321,6 +343,7 @@ async fn run() -> Result<()> {
base_provider,
)
.inspect_err(|error| report_init_failure(error, None))?;
startup_trace!("rsi provider wrapped");
let provider = rsi_bundle.provider;
let rsi_control = rsi_bundle.rsi_control;
let rsi_remote_switch = rsi_bundle.rsi_remote_switch;
Expand Down Expand Up @@ -393,6 +416,7 @@ async fn run() -> Result<()> {
}
};

startup_trace!("agent built");
let managed_context = cli
.rsi_context_json
.as_deref()
Expand Down Expand Up @@ -431,8 +455,10 @@ async fn run() -> Result<()> {
// When sync is on, also create a RemoteUi for live event streaming.
// Clone the path before it's moved into JsonlSession — the daemon fallback
// below may need to create its own session sink.
startup_trace!("delegate runner ready");
let daemon_session_path = session_path.clone();
let sync_store = sync_store::SyncStore::open()?;
startup_trace!("sync store opened");
let legacy_enabled = file.sync.as_ref().is_some_and(|section| section.enabled);
let mut persisted_sync_mode = sync_store.initialize_mode(legacy_enabled)?;
if let Some(configured) = file.sync.as_ref().and_then(|section| section.mode) {
Expand All @@ -454,7 +480,9 @@ async fn run() -> Result<()> {
.clone()
.unwrap_or_else(|| feedback::session_id_from_path(&session_path));
let remote = sync::RemoteSessionSink::new(sync_config.clone(), session_id.clone());
startup_trace!("remote sync sink built");
let sync_session = sync::SyncSession::new(JsonlSession::new(session_path), remote);
startup_trace!("sync session reconciled");
let handle = sync_session.remote_handle();
agent.set_session(Box::new(sync_session));
let remote_ui = std::sync::Arc::new(sync::RemoteUi::new(sync_config, session_id));
Expand Down Expand Up @@ -745,17 +773,17 @@ async fn run() -> Result<()> {
}
// A one-shot turn may have started background processes; don't leak them.
agent.kill_background_processes();
// Flush any pending sync records and live events to ipop before exiting.
// Flush any pending sync records and live events to ipop before
// exiting. Silent on failure by design: sync is best-effort mirroring
// of a local-first session — everything unsent stays queued in the
// durable outbox for a later process, and a portal outage must never
// surface as an error in the coding workflow.
if let Some(handle) = &sync_handle {
if let Err(err) = handle.flush().await {
eprintln!("\x1b[33msync: {err:#}\x1b[0m");
}
let _ = handle.flush().await;
handle.end_session().await;
}
if let Some(rui) = &remote_ui
&& let Err(err) = rui.flush().await
{
eprintln!("\x1b[33msync events: {err:#}\x1b[0m");
if let Some(rui) = &remote_ui {
let _ = rui.flush().await;
}
if report_result.is_err() {
std::process::exit(3);
Expand Down Expand Up @@ -1027,8 +1055,12 @@ async fn run() -> Result<()> {
remote.seed_snapshot(&loaded)?;
// Stage the replacement completely, including the
// automatic takeover lease, before touching the
// live agent or persistence handles.
remote.ensure_registered_now().await?;
// live agent or persistence handles. Sync is
// best-effort: an unreachable portal must not
// fail the local session switch — and must not
// surface as an error either. Registration is
// retried by later flushes.
let _ = remote.ensure_registered_now_quiet().await;
let synced =
sync::SyncSession::new(JsonlSession::new(path.clone()), remote);
let next_handle = synced.remote_handle();
Expand Down Expand Up @@ -1166,6 +1198,7 @@ async fn run() -> Result<()> {
}),
purge: std::sync::Arc::new(|| sync_store::SyncStore::open()?.purge()),
};
startup_trace!("handing off to TUI");
match hi_tui::run(
&mut agent,
hi_tui::RunOptions {
Expand Down Expand Up @@ -1202,18 +1235,17 @@ async fn run() -> Result<()> {
Ok(()) => {
let active_session_id = tui_active_session_id.lock().unwrap().clone();
feedback::maybe_prompt_and_submit(&settings, &active_session_id).await;
// Best-effort exit flush: anything unsent stays in the
// durable outbox, and portal trouble never surfaces as an
// error in the coding workflow.
let active_handle = tui_sync_handle.lock().unwrap().clone();
if let Some(handle) = &active_handle {
if let Err(err) = handle.flush().await {
eprintln!("\x1b[33msync: {err:#}\x1b[0m");
}
let _ = handle.flush().await;
handle.end_session().await;
}
let active_remote_ui = tui_remote_ui.lock().unwrap().clone();
if let Some(rui) = &active_remote_ui
&& let Err(err) = rui.flush().await
{
eprintln!("\x1b[33msync events: {err:#}\x1b[0m");
if let Some(rui) = &active_remote_ui {
let _ = rui.flush().await;
}
agent.kill_background_processes();
finish_interactive_trace(rsi.observer.as_ref(), &agent)?;
Expand Down Expand Up @@ -1274,16 +1306,13 @@ async fn run() -> Result<()> {
if repl_result.is_ok() {
feedback::maybe_prompt_and_submit(&settings, &feedback_session_id).await;
}
// Best-effort exit flush: silent by design — see the TUI exit path.
if let Some(handle) = &sync_handle {
if let Err(err) = handle.flush().await {
eprintln!("\x1b[33msync: {err:#}\x1b[0m");
}
let _ = handle.flush().await;
handle.end_session().await;
}
if let Some(rui) = &remote_ui
&& let Err(err) = rui.flush().await
{
eprintln!("\x1b[33msync events: {err:#}\x1b[0m");
if let Some(rui) = &remote_ui {
let _ = rui.flush().await;
}
finish_interactive_trace(rsi.observer.as_ref(), &agent)?;
repl_result
Expand Down
Loading
Loading