diff --git a/src/function/jobs.rs b/src/function/jobs.rs index b2fd2f5..f2d6c92 100644 --- a/src/function/jobs.rs +++ b/src/function/jobs.rs @@ -479,6 +479,7 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result { state, output_buf, no_change_checks: 0, + last_check_state: None, }; // On a capacity race the handle is dropped here, which kills the process @@ -507,8 +508,8 @@ fn handle_check(ctx: &RequestContext, args: &Value) -> Result { let Some(supervisor) = ctx.supervisor.as_ref() else { return Ok(job_miss_error(None, id)); }; - let sup = supervisor.read(); - let Some(job) = sup.job(id) else { + let mut sup = supervisor.write(); + let Some(job) = sup.job_mut(id) else { drop(sup); return Ok(job_miss_error(ctx.supervisor.as_ref(), id)); }; @@ -518,6 +519,13 @@ fn handle_check(ctx: &RequestContext, args: &Value) -> Result { let buf = job.output_buf.lock(); (buf.tail(), buf.total_written()) }; + let check_state = (status, total_written); + if job.last_check_state == Some(check_state) { + job.no_change_checks += 1; + } else { + job.no_change_checks = 0; + job.last_check_state = Some(check_state); + } let tail_truncated = (tail.len() as u64) < total_written; let mut result = json!({ "status": job_status_str(status), @@ -532,6 +540,11 @@ fn handle_check(ctx: &RequestContext, args: &Value) -> Result { result["message"] = json!( "Job is still running. Call job__collect to block for the result, or do other work — you will be notified on completion." ); + if job.no_change_checks >= 3 { + result["hint"] = json!( + "No change across repeated checks — still running; call job__collect to block, or do other work — a system notification will fire on completion." + ); + } } else { result["message"] = json!(format!( "Job finished — retrieve the result with job__collect --id {id}" @@ -1175,6 +1188,7 @@ mod tests { })), output_buf: Arc::new(Mutex::new(RingBuf::default())), no_change_checks: 0, + last_check_state: None, } } @@ -1640,6 +1654,7 @@ mod tests { })), output_buf: Arc::new(Mutex::new(RingBuf::default())), no_change_checks: 0, + last_check_state: None, }; ctx.supervisor .as_ref() @@ -1754,6 +1769,86 @@ mod tests { assert!(ctx.supervisor.as_ref().unwrap().read().has_job("j1")); } + #[test] + fn handle_check_hints_after_repeated_unchanged_checks() { + let ctx = ctx_with_job_supervisor(4); + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(make_running_job("j1")) + .unwrap(); + + for _ in 0..3 { + let result = handle_check(&ctx, &json!({"id": "j1"})).unwrap(); + assert!(result.get("hint").is_none()); + } + let result = handle_check(&ctx, &json!({"id": "j1"})).unwrap(); + assert_eq!( + result["hint"], + "No change across repeated checks — still running; call job__collect to block, or do other work — a system notification will fire on completion." + ); + } + + #[test] + fn handle_check_no_change_counter_resets_on_output_change() { + let ctx = ctx_with_job_supervisor(4); + let job = make_running_job("j1"); + let output_buf = Arc::clone(&job.output_buf); + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(job) + .unwrap(); + + for _ in 0..3 { + handle_check(&ctx, &json!({"id": "j1"})).unwrap(); + } + assert!( + handle_check(&ctx, &json!({"id": "j1"})) + .unwrap() + .get("hint") + .is_some() + ); + + output_buf.lock().push(b"more"); + for _ in 0..3 { + let result = handle_check(&ctx, &json!({"id": "j1"})).unwrap(); + assert!(result.get("hint").is_none()); + } + assert!( + handle_check(&ctx, &json!({"id": "j1"})) + .unwrap() + .get("hint") + .is_some() + ); + } + + #[test] + fn handle_check_finished_job_never_hints() { + let ctx = ctx_with_job_supervisor(4); + let job = make_running_job("j1"); + job.state.lock().status = JobStatus::Completed; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(job) + .unwrap(); + + for _ in 0..5 { + let result = handle_check(&ctx, &json!({"id": "j1"})).unwrap(); + assert!(result.get("hint").is_none()); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("job__collect --id j1") + ); + } + } + #[test] fn handle_collect_applies_tail_lines() { run_async(async { @@ -1777,6 +1872,7 @@ mod tests { })), output_buf: Arc::new(Mutex::new(RingBuf::default())), no_change_checks: 0, + last_check_state: None, }; ctx.supervisor .as_ref() @@ -1815,6 +1911,7 @@ mod tests { })), output_buf, no_change_checks: 0, + last_check_state: None, }; ctx.supervisor .as_ref() @@ -1862,6 +1959,7 @@ mod tests { state: Arc::clone(&state), output_buf, no_change_checks: 0, + last_check_state: None, }; ctx.supervisor .as_ref() @@ -2115,6 +2213,7 @@ mod tests { state: Arc::clone(&state), output_buf, no_change_checks: 0, + last_check_state: None, }; ctx.supervisor .as_ref() diff --git a/src/function/mod.rs b/src/function/mod.rs index 5b1b614..e116dd0 100644 --- a/src/function/mod.rs +++ b/src/function/mod.rs @@ -2409,6 +2409,19 @@ fn polyfill_cmd_name>(cmd_name: &str, bin_dir: &[T]) -> String { cmd_name } +// Polling tools are expected to repeat; recording them would also let them +// break up detection of a real loop in the calls they interleave with. +const LOOP_TRACKER_EXEMPT_TOOLS: [&str; 4] = [ + "job__check", + "job__list", + "agent__check", + "agent__list_running", +]; + +fn is_loop_tracker_exempt(name: &str) -> bool { + LOOP_TRACKER_EXEMPT_TOOLS.contains(&name) +} + #[derive(Debug, Clone)] pub struct ToolCallTracker { last_calls: VecDeque, @@ -2430,6 +2443,9 @@ impl ToolCallTracker { } pub fn check_loop(&self, new_call: &ToolCall) -> Option { + if is_loop_tracker_exempt(&new_call.name) { + return None; + } if self.last_calls.len() < self.max_repeats { return None; } @@ -2496,6 +2512,9 @@ impl ToolCallTracker { } pub fn record_call(&mut self, call: ToolCall) { + if is_loop_tracker_exempt(&call.name) { + return; + } if self.last_calls.len() >= self.chain_len * self.max_repeats { self.last_calls.pop_front(); } @@ -2663,6 +2682,7 @@ mod tests { })), output_buf: Arc::new(parking_lot::Mutex::new(jobs::RingBuf::default())), no_change_checks: 0, + last_check_state: None, }; let mut sup = crate::supervisor::Supervisor::new(0, 3).with_max_concurrent_jobs(4); sup.register(handle).unwrap(); @@ -3105,6 +3125,46 @@ mod tests { assert!(msg.contains("repeat_tool")); } + #[test] + fn tracker_exempt_tools_never_trip() { + for name in LOOP_TRACKER_EXEMPT_TOOLS { + let mut tracker = ToolCallTracker::default(); + let exempt = call_with_args(name, json!({"id": "j1"})); + tracker.record_call(exempt.clone()); + tracker.record_call(exempt.clone()); + assert!(tracker.check_loop(&exempt).is_none()); + + let other = call_with_args("execute_command", json!({"command": "ls"})); + tracker.record_call(other.clone()); + assert!( + tracker.check_loop(&other).is_none(), + "exempt calls must not count toward the repeat threshold" + ); + tracker.record_call(other.clone()); + assert!(tracker.check_loop(&other).is_some()); + } + } + + #[test] + fn tracker_exempt_interleave_does_not_mask_real_loop() { + let mut tracker = ToolCallTracker::default(); + let x = call_with_args("execute_command", json!({"command": "ls"})); + tracker.record_call(call_with_args("job__check", json!({"id": "j1"}))); + tracker.record_call(x.clone()); + tracker.record_call(call_with_args("job__check", json!({"id": "j1"}))); + tracker.record_call(x.clone()); + assert!(tracker.check_loop(&x).is_some()); + } + + #[test] + fn tracker_non_exempt_behavior_unchanged() { + let mut tracker = ToolCallTracker::default(); + let c = call_with_args("fs_cat", json!({"path": "a.txt"})); + tracker.record_call(c.clone()); + tracker.record_call(c.clone()); + assert!(tracker.check_loop(&c).is_some()); + } + #[test] fn prefix_constants_are_correct() { assert_eq!(TODO_FUNCTION_PREFIX, "todo__"); diff --git a/src/function/supervisor.rs b/src/function/supervisor.rs index fee6e3e..e1586ab 100644 --- a/src/function/supervisor.rs +++ b/src/function/supervisor.rs @@ -1635,6 +1635,7 @@ mod tests { })), output_buf: Arc::new(Mutex::new(RingBuf::default())), no_change_checks: 0, + last_check_state: None, } } diff --git a/src/supervisor/mod.rs b/src/supervisor/mod.rs index 05ececb..ef8e2fc 100644 --- a/src/supervisor/mod.rs +++ b/src/supervisor/mod.rs @@ -67,8 +67,8 @@ pub struct JobHandle { pub abort_signal: AbortSignal, pub state: Arc>, pub output_buf: Arc>, - #[allow(dead_code)] pub no_change_checks: u32, + pub last_check_state: Option<(JobStatus, u64)>, } impl JobHandle { @@ -159,6 +159,13 @@ impl Supervisor { } } + pub fn job_mut(&mut self, id: &str) -> Option<&mut JobHandle> { + match self.handles.get_mut(id) { + Some(TaskHandle::Job(handle)) => Some(handle), + _ => None, + } + } + pub fn jobs(&self) -> impl Iterator { self.handles.values().filter_map(|handle| match handle { TaskHandle::Job(handle) => Some(handle), @@ -392,6 +399,7 @@ mod tests { })), output_buf: Arc::new(Mutex::new(RingBuf::default())), no_change_checks: 0, + last_check_state: None, } }