Skip to content

fix(cortex): use idle time instead of lifetime for worker timeout #332

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
jamiepine merged 2 commits into from
Mar 5, 2026
Merged

fix(cortex): use idle time instead of lifetime for worker timeout #332

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e24089b
fix(cortex): use idle time instead of lifetime for worker timeout
jamiepine Mar 5, 2026
c119f8b
fix(cortex): bias select! to drain events before health tick
jamiepine Mar 5, 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
4 changes: 2 additions & 2 deletions docs/content/docs/(configuration)/config.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,7 @@ emergency_threshold = 0.95 # drop oldest 50%, no LLM
# Cortex (system observer) settings.
[defaults.cortex]
tick_interval_secs = 30
worker_timeout_secs = 300
worker_timeout_secs = 600
branch_timeout_secs = 60
circuit_breaker_threshold = 3 # consecutive failures before auto-disable

Expand Down Expand Up @@ -504,7 +504,7 @@ Thresholds are fractions of `context_window`.
| Key | Type | Default | Description |
|-----|------|---------|-------------|
| `tick_interval_secs` | integer | 30 | How often the cortex runtime loop runs maintenance ticks while continuously observing events |
| `worker_timeout_secs` | integer | 300 | Worker timeout before cancellation |
| `worker_timeout_secs` | integer | 600 | Worker idle timeout before cancellation |
| `branch_timeout_secs` | integer | 60 | Branch timeout before cancellation |
| `detached_worker_timeout_retry_limit` | integer | 2 | Retry limit before quarantining detached workers to backlog |
| `supervisor_kill_budget_per_tick` | integer | 8 | Max number of overdue processes supervisor may cancel per health tick |
Expand Down
4 changes: 2 additions & 2 deletions docs/content/docs/(core)/cortex.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,8 +164,8 @@ bulletin_interval_secs = 3600
# Target word count for the memory bulletin.
bulletin_max_words = 500

# Worker is considered hanging if no status update for this long.
worker_timeout_secs = 300
# Worker is considered hanging if no activity for this long.
worker_timeout_secs = 600

# Branch is considered stale after this duration.
branch_timeout_secs = 60
Expand Down
95 changes: 81 additions & 14 deletions src/agent/cortex.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,6 +258,7 @@ struct WorkerTracker {
channel_id: Option<ChannelId>,
worker_type: String,
started_at: Instant,
last_activity_at: Instant,
}

#[derive(Debug, Clone)]
Expand Down Expand Up @@ -302,17 +303,25 @@ impl HealthRuntimeState {
channel_id: Option<ChannelId>,
worker_type: String,
) {
let now = Instant::now();
self.worker_trackers.insert(
worker_id,
WorkerTracker {
worker_id,
channel_id,
worker_type,
started_at: Instant::now(),
started_at: now,
last_activity_at: now,
},
);
}

fn track_worker_activity(&mut self, worker_id: WorkerId) {
if let Some(tracker) = self.worker_trackers.get_mut(&worker_id) {
tracker.last_activity_at = Instant::now();
}
}

fn track_worker_complete(&mut self, worker_id: WorkerId, success: bool, threshold: u8) {
let Some(worker_type) = self
.worker_trackers
Expand Down Expand Up @@ -398,9 +407,9 @@ fn parse_structured_success_flag(result: &str) -> Option<bool> {
object.get("ok").and_then(|value| value.as_bool())
}

fn kill_target_started_at(target: &KillTarget) -> Instant {
fn kill_target_last_activity(target: &KillTarget) -> Instant {
match target {
KillTarget::Worker(tracker) => tracker.started_at,
KillTarget::Worker(tracker) => tracker.last_activity_at,
KillTarget::Branch(tracker) => tracker.started_at,
}
}
Expand All @@ -420,12 +429,12 @@ fn build_kill_targets(
targets.extend(overdue_workers.into_iter().map(KillTarget::Worker));
targets.extend(overdue_branches.into_iter().map(KillTarget::Branch));
targets.sort_by(|left, right| {
let left_started = kill_target_started_at(left);
let right_started = kill_target_started_at(right);
if left_started == right_started {
let left_activity = kill_target_last_activity(left);
let right_activity = kill_target_last_activity(right);
if left_activity == right_activity {
kill_target_id(left).cmp(&kill_target_id(right))
} else {
left_started.cmp(&right_started)
left_activity.cmp(&right_activity)
}
});
targets
Expand Down Expand Up @@ -819,15 +828,32 @@ impl Cortex {
ProcessEvent::WorkerComplete {
worker_id, success, ..
} => state.track_worker_complete(*worker_id, *success, threshold),
ProcessEvent::WorkerStatus { worker_id, .. } => {

@tembo tembo Bot Mar 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: should ProcessEvent::StatusUpdate { process_id: ProcessId::Worker(..) } also reset the idle clock? SpacebotHook::send_status emits StatusUpdate, so those updates won’t count as activity here.

Suggested change
ProcessEvent::WorkerStatus { worker_id, .. } => {
ProcessEvent::StatusUpdate {
process_id: ProcessId::Worker(worker_id),
..
} => state.track_worker_activity(*worker_id),

state.track_worker_activity(*worker_id);
}
ProcessEvent::ToolStarted {
process_id: ProcessId::Worker(worker_id),
..
} => {
state.track_worker_activity(*worker_id);
}
ProcessEvent::ToolCompleted {
process_id,
tool_name,
result,
..
} => {
if let ProcessId::Worker(worker_id) = process_id {
state.track_worker_activity(*worker_id);
}
state.track_tool_completed(tool_name, result, threshold);
}
ProcessEvent::BranchStarted {
branch_id,
channel_id,
..
} => state.track_branch_start(*branch_id, channel_id.clone()),
ProcessEvent::BranchResult { branch_id, .. } => state.track_branch_complete(*branch_id),
ProcessEvent::ToolCompleted {
tool_name, result, ..
} => state.track_tool_completed(tool_name, result, threshold),
_ => {}
}
}
Expand Down Expand Up @@ -866,7 +892,9 @@ impl Cortex {
state
.worker_trackers
.values()
.filter(|tracker| now.duration_since(tracker.started_at) >= worker_timeout)
.filter(|tracker| {
now.duration_since(tracker.last_activity_at) >= worker_timeout
})
Comment thread
coderabbitai[bot] marked this conversation as resolved.
.cloned()
.collect()
};
Expand Down Expand Up @@ -927,8 +955,12 @@ impl Cortex {
kill_attempts = kill_attempts.saturating_add(1);
let result = match target.clone() {
KillTarget::Worker(tracker) => {
let reason =
format!("timed out after {}s (supervisor)", worker_timeout.as_secs());
let idle_secs = now.duration_since(tracker.last_activity_at).as_secs();
let reason = format!(
"idle for {}s, exceeded {}s timeout (supervisor)",
idle_secs,
worker_timeout.as_secs()
);
if let Some(channel_id) = &tracker.channel_id {
self.deps
.process_control_registry
Expand Down Expand Up @@ -959,14 +991,18 @@ impl Cortex {
KillTarget::Worker(tracker) => {
terminal_worker_ids.push(tracker.worker_id);
if is_cancelled_control_result(result) {
let idle_secs = now.duration_since(tracker.last_activity_at).as_secs();
let lifetime_secs = now.duration_since(tracker.started_at).as_secs();
logger.log(
"worker_killed",
&format!("Worker {} cancelled by supervisor", tracker.worker_id),
Some(serde_json::json!({
"worker_id": tracker.worker_id.to_string(),
"channel_id": tracker.channel_id.as_deref(),
"idle_secs": idle_secs,
"lifetime_secs": lifetime_secs,
"timeout_secs": worker_timeout.as_secs(),
"reason": "timeout",
"reason": "idle_timeout",
})),
);
kill_actions = kill_actions.saturating_add(1);
Expand Down Expand Up @@ -3958,13 +3994,15 @@ mod tests {
channel_id: Some(Arc::from("channel-a")),
worker_type: "builtin".to_string(),
started_at: shared_start,
last_activity_at: shared_start,
};
let worker_b = WorkerTracker {
worker_id: uuid::Uuid::parse_str("00000000-0000-0000-0000-00000000000b")
.expect("valid uuid"),
channel_id: Some(Arc::from("channel-a")),
worker_type: "builtin".to_string(),
started_at: shared_start,
last_activity_at: shared_start,
};
let branch_oldest = BranchTracker {
branch_id: uuid::Uuid::parse_str("00000000-0000-0000-0000-000000000001")
Expand Down Expand Up @@ -4003,6 +4041,35 @@ mod tests {
);
}

#[test]
fn worker_activity_resets_idle_clock() {
let mut state = HealthRuntimeState::default();
let worker_id = uuid::Uuid::new_v4();
state.track_worker_start(worker_id, Some(Arc::from("ch")), "builtin".to_string());

let tracker_before = state.worker_trackers.get(&worker_id).unwrap().clone();
// Simulate time passing by checking that activity updates the timestamp.
std::thread::sleep(std::time::Duration::from_millis(10));
state.track_worker_activity(worker_id);

let tracker_after = state.worker_trackers.get(&worker_id).unwrap();
assert!(
tracker_after.last_activity_at > tracker_before.last_activity_at,
"last_activity_at should advance after track_worker_activity"
);
assert_eq!(
tracker_after.started_at, tracker_before.started_at,
"started_at should not change"
);
}
Comment on lines +4046 to +4065

@tembo tembo Bot Mar 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unit test: avoiding thread::sleep here should make this less timing-sensitive/flaky on slow CI. You can force an older last_activity_at and assert it advances.

Suggested change
fn worker_activity_resets_idle_clock() {
let mut state = HealthRuntimeState::default();
let worker_id = uuid::Uuid::new_v4();
state.track_worker_start(worker_id, Some(Arc::from("ch")), "builtin".to_string());
let tracker_before = state.worker_trackers.get(&worker_id).unwrap().clone();
// Simulate time passing by checking that activity updates the timestamp.
std::thread::sleep(std::time::Duration::from_millis(10));
state.track_worker_activity(worker_id);
let tracker_after = state.worker_trackers.get(&worker_id).unwrap();
assert!(
tracker_after.last_activity_at > tracker_before.last_activity_at,
"last_activity_at should advance after track_worker_activity"
);
assert_eq!(
tracker_after.started_at, tracker_before.started_at,
"started_at should not change"
);
}
fn worker_activity_resets_idle_clock() {
let mut state = HealthRuntimeState::default();
let worker_id = uuid::Uuid::new_v4();
state.track_worker_start(worker_id, Some(Arc::from("ch")), "builtin".to_string());
let started_at = state.worker_trackers.get(&worker_id).unwrap().started_at;
let previous_activity_at = {
let tracker = state.worker_trackers.get_mut(&worker_id).unwrap();
tracker.last_activity_at = tracker
.last_activity_at
.checked_sub(Duration::from_secs(1))
.unwrap();
tracker.last_activity_at
};
state.track_worker_activity(worker_id);
let tracker_after = state.worker_trackers.get(&worker_id).unwrap();
assert!(
tracker_after.last_activity_at > previous_activity_at,
"last_activity_at should advance after track_worker_activity"
);
assert_eq!(
tracker_after.started_at, started_at,
"started_at should not change"
);
}


#[test]
fn worker_activity_noop_for_unknown_worker() {
let mut state = HealthRuntimeState::default();
// Should not panic on unknown worker ID.
state.track_worker_activity(uuid::Uuid::new_v4());
}

#[test]
fn terminal_control_result_includes_not_found_and_already_terminal() {
assert!(is_terminal_control_result(ControlActionResult::Cancelled));
Expand Down
2 changes: 1 addition & 1 deletion src/config/types.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -835,7 +835,7 @@ impl Default for CortexConfig {
fn default() -> Self {
Self {
tick_interval_secs: 30,
worker_timeout_secs: 300,
worker_timeout_secs: 600,
branch_timeout_secs: 60,
detached_worker_timeout_retry_limit: 2,
supervisor_kill_budget_per_tick: 8,
Expand Down