Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
82 changes: 77 additions & 5 deletions crates/openfang-hands/src/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,12 @@ impl HandRegistry {
.map(|e| {
serde_json::json!({
"hand_id": e.hand_id,
// Without the name, a named instance is restored as an
// unnamed one: it takes the legacy agent id derived from
// hand_id, is renamed to the hand's default, and lands in a
// different workspace — orphaning the context.md and files
// the original instance had accumulated.
"instance_name": e.instance_name,
"config": e.config,
"agent_id": e.agent_id,
})
Expand All @@ -123,12 +129,24 @@ impl HandRegistry {
}

/// Load persisted hand state and re-activate hands.
/// Returns list of (hand_id, config, old_agent_id) that should be activated.
/// The `old_agent_id` is the agent UUID from before the restart, used to
///
/// Returns `(hand_id, instance_name, config, old_agent_id)` for each hand
/// that should be activated. `instance_name` must be carried through:
/// activating without it derives a different agent id and a different agent
/// name, so a named instance would not survive a restart. It is optional in
/// the file so state written before this field existed still loads.
///
/// `old_agent_id` is the agent UUID from before the restart, used to
/// reassign cron jobs to the newly spawned agent (issue #402).
#[allow(clippy::type_complexity)]
pub fn load_state(
path: &std::path::Path,
) -> Vec<(String, HashMap<String, serde_json::Value>, Option<AgentId>)> {
) -> Vec<(
String,
Option<String>,
HashMap<String, serde_json::Value>,
Option<AgentId>,
)> {
let data = match std::fs::read_to_string(path) {
Ok(d) => d,
Err(_) => return Vec::new(),
Expand All @@ -144,12 +162,16 @@ impl HandRegistry {
.into_iter()
.filter_map(|e| {
let hand_id = e["hand_id"].as_str()?.to_string();
let instance_name = e
.get("instance_name")
.and_then(|v| v.as_str())
.map(str::to_string);
let config: HashMap<String, serde_json::Value> =
serde_json::from_value(e["config"].clone()).unwrap_or_default();
let old_agent_id: Option<AgentId> = e
.get("agent_id")
.and_then(|v| serde_json::from_value(v.clone()).ok());
Some((hand_id, config, old_agent_id))
Some((hand_id, instance_name, config, old_agent_id))
})
.collect()
}
Expand Down Expand Up @@ -1355,11 +1377,61 @@ system_prompt = "v2 — schedule changed."
reg.persist_state(&state_file).unwrap();
let reloaded = HandRegistry::load_state(&state_file);
assert_eq!(reloaded.len(), 1);
let (hand_id, config, _agent_id) = &reloaded[0];
let (hand_id, _instance_name, config, _agent_id) = &reloaded[0];
assert_eq!(hand_id, "browser");
assert_eq!(
config.get("headless"),
Some(&serde_json::Value::String("true".into()))
);
}

/// A named instance must come back named.
///
/// The name decides the agent id (`hand_instance_<uuid>` vs the legacy
/// id derived from `hand_id`), the agent name, and therefore the workspace
/// directory. Dropping it on restore silently re-homed the instance and
/// orphaned everything it had accumulated there.
#[test]
fn test_instance_name_survives_persist_reload() {
let reg = test_registry_with_dummy_hand("collector");
let inst = reg
.activate("collector", HashMap::new(), Some("underground".into()))
.unwrap();
assert_eq!(inst.instance_name.as_deref(), Some("underground"));

let tmp = tempfile::tempdir().unwrap();
let state_file = tmp.path().join("hands.json");
reg.persist_state(&state_file).unwrap();

let reloaded = HandRegistry::load_state(&state_file);
assert_eq!(reloaded.len(), 1);
let (hand_id, instance_name, _config, _agent_id) = &reloaded[0];
assert_eq!(hand_id, "collector");
assert_eq!(
instance_name.as_deref(),
Some("underground"),
"named instance must be restored under the same name"
);
}

/// State written before `instance_name` existed must still load.
#[test]
fn test_load_state_without_instance_name_is_unnamed() {
let tmp = tempfile::tempdir().unwrap();
let state_file = tmp.path().join("hands.json");
std::fs::write(
&state_file,
r#"[{"hand_id":"collector","config":{},"agent_id":null}]"#,
)
.unwrap();

let reloaded = HandRegistry::load_state(&state_file);
assert_eq!(reloaded.len(), 1);
let (hand_id, instance_name, _config, _agent_id) = &reloaded[0];
assert_eq!(hand_id, "collector");
assert!(
instance_name.is_none(),
"legacy state has no name and must load as unnamed"
);
}
}
83 changes: 67 additions & 16 deletions crates/openfang-kernel/src/heartbeat.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,17 @@ const DEFAULT_CHECK_INTERVAL_SECS: u64 = 30;
/// multiples of its heartbeat interval.
const UNRESPONSIVE_MULTIPLIER: u64 = 2;

/// Seconds of inactivity before a Running agent is presumed unresponsive.
///
/// Must not be shorter than the longest a legitimate turn may take, or live
/// work gets killed and restarted: an agent still inside its turn budget is
/// waiting, not hung. `kernel::DEFAULT_AGENT_TURN_TIMEOUT_SECS` is 600, and
/// per-job `timeout_secs` is bounded by `scheduler::MAX_TIMEOUT_SECS` (3600),
/// so 600 matches the default turn and the turn timeout remains the mechanism
/// that ends an over-running turn. This value only decides when to call an
/// agent *dead*.
const DEFAULT_UNRESPONSIVE_TIMEOUT_SECS: u64 = 600;

/// Default maximum recovery attempts before giving up.
const DEFAULT_MAX_RECOVERY_ATTEMPTS: u32 = 3;

Expand Down Expand Up @@ -61,8 +72,7 @@ impl Default for HeartbeatConfig {
fn default() -> Self {
Self {
check_interval_secs: DEFAULT_CHECK_INTERVAL_SECS,
// 180s default: browser tasks and complex LLM calls can take 1-3 minutes
default_timeout_secs: 180,
default_timeout_secs: DEFAULT_UNRESPONSIVE_TIMEOUT_SECS,
max_recovery_attempts: DEFAULT_MAX_RECOVERY_ATTEMPTS,
recovery_cooldown_secs: DEFAULT_RECOVERY_COOLDOWN_SECS,
}
Expand Down Expand Up @@ -157,12 +167,28 @@ pub fn check_agents(registry: &AgentRegistry, config: &HeartbeatConfig) -> Vec<H

let inactive_secs = (now - entry_ref.last_active).num_seconds();

// Determine timeout: use agent's autonomous config if set, else default
// Determine timeout: use agent's autonomous config if set, else default.
//
// The per-agent value is a FLOOR-RAISER, never a floor-lowerer. Taken
// literally it produces a *tighter* bound than the default for exactly
// the agents doing the heaviest work: the default heartbeat interval is
// 30s, so `30 * 2 = 60s` — well under the 180s the default already
// concedes that "browser tasks and complex LLM calls" need, and far
// under the 600s a single agent turn is allowed to run for.
//
// A collector agent waiting on a rate-limited reasoning model was being
// declared crashed every 60 seconds and "recovered" — restarting its
// loop and discarding the in-flight turn, forever. Waiting on the model
// is not being hung. Take the larger of the two bounds so an autonomous
// config can only ever grant an agent MORE time, not less.
let timeout_secs = entry_ref
.manifest
.autonomous
.as_ref()
.map(|a| a.heartbeat_interval_secs * UNRESPONSIVE_MULTIPLIER)
.map(|a| {
(a.heartbeat_interval_secs * UNRESPONSIVE_MULTIPLIER)
.max(config.default_timeout_secs)
})
.unwrap_or(config.default_timeout_secs) as i64;

// --- Skip idle agents that have never genuinely processed a message ---
Expand Down Expand Up @@ -382,7 +408,7 @@ mod tests {
);
registry.register(idle_agent).unwrap();

let config = HeartbeatConfig::default(); // timeout = 180s
let config = HeartbeatConfig::default();
let statuses = check_agents(&registry, &config);

// The idle agent should be skipped entirely
Expand Down Expand Up @@ -460,17 +486,15 @@ mod tests {
// An agent that WAS active (last_active >> created_at) but has gone
// silent for longer than the timeout — should be flagged unresponsive.
let registry = crate::registry::AgentRegistry::new();
let ten_min_ago = Utc::now() - Duration::seconds(600);
let five_min_ago = Utc::now() - Duration::seconds(300);
let active_agent = make_entry(
"active-agent",
AgentState::Running,
ten_min_ago,
five_min_ago,
);
// Derive the timings from the threshold rather than hardcoding, so
// this keeps testing "past the timeout" if the default ever moves.
let timeout = DEFAULT_UNRESPONSIVE_TIMEOUT_SECS as i64;
let created = Utc::now() - Duration::seconds(timeout * 3);
let last_active = Utc::now() - Duration::seconds(timeout + 60);
let active_agent = make_entry("active-agent", AgentState::Running, created, last_active);
registry.register(active_agent).unwrap();

let config = HeartbeatConfig::default(); // timeout = 180s, inactive = ~300s
let config = HeartbeatConfig::default();
let statuses = check_agents(&registry, &config);

assert_eq!(statuses.len(), 1);
Expand All @@ -489,7 +513,7 @@ mod tests {
let healthy_agent = make_entry("healthy-agent", AgentState::Running, ten_min_ago, just_now);
registry.register(healthy_agent).unwrap();

let config = HeartbeatConfig::default(); // timeout = 180s
let config = HeartbeatConfig::default();
let statuses = check_agents(&registry, &config);

assert_eq!(statuses.len(), 1);
Expand Down Expand Up @@ -544,7 +568,34 @@ mod tests {
fn test_heartbeat_config_default() {
let config = HeartbeatConfig::default();
assert_eq!(config.check_interval_secs, 30);
assert_eq!(config.default_timeout_secs, 180);
assert_eq!(
config.default_timeout_secs,
DEFAULT_UNRESPONSIVE_TIMEOUT_SECS
);
}

/// An autonomous config may only grant an agent MORE time before it is
/// presumed dead, never less.
///
/// Read literally, `heartbeat_interval_secs * UNRESPONSIVE_MULTIPLIER` is
/// 60s for the default 30s interval — tighter than the default timeout, so
/// an agent waiting on a slow model was declared crashed and had its
/// in-flight turn restarted every 60 seconds.
#[test]
fn test_autonomous_config_never_shortens_timeout() {
let config = HeartbeatConfig::default();
let effective =
|interval: u64| (interval * UNRESPONSIVE_MULTIPLIER).max(config.default_timeout_secs);

// The default 30s interval must not drag the bound below the default.
assert_eq!(effective(30), config.default_timeout_secs);
assert!(
effective(30) >= config.default_timeout_secs,
"a short heartbeat interval must not shorten the death threshold"
);

// A generous interval still raises it.
assert_eq!(effective(600), 1200);
}

#[test]
Expand Down
9 changes: 7 additions & 2 deletions crates/openfang-kernel/src/kernel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4414,8 +4414,13 @@ impl OpenFangKernel {
let saved_hands = openfang_hands::registry::HandRegistry::load_state(&state_path);
if !saved_hands.is_empty() {
info!("Restoring {} persisted hand(s)", saved_hands.len());
for (hand_id, config, old_agent_id) in saved_hands {
match self.activate_hand(&hand_id, config, None) {
for (hand_id, instance_name, config, old_agent_id) in saved_hands {
// Restore under the SAME instance name. Passing None here made
// every named instance come back as the unnamed one: a
// different agent id, the hand's default agent name, and a
// different workspace — so its context.md and accumulated files
// were orphaned on every restart.
match self.activate_hand(&hand_id, config, instance_name) {
Ok(inst) => {
info!(hand = %hand_id, instance = %inst.instance_id, "Hand restored");
// Reassign cron jobs and triggers from the pre-restart
Expand Down