feat(jobs): exempt polling tools from loop tracker and hint on unchanged checks
This commit is contained in:
+101
-2
@@ -479,6 +479,7 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
|
|||||||
state,
|
state,
|
||||||
output_buf,
|
output_buf,
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
};
|
};
|
||||||
|
|
||||||
// On a capacity race the handle is dropped here, which kills the process
|
// 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<Value> {
|
|||||||
let Some(supervisor) = ctx.supervisor.as_ref() else {
|
let Some(supervisor) = ctx.supervisor.as_ref() else {
|
||||||
return Ok(job_miss_error(None, id));
|
return Ok(job_miss_error(None, id));
|
||||||
};
|
};
|
||||||
let sup = supervisor.read();
|
let mut sup = supervisor.write();
|
||||||
let Some(job) = sup.job(id) else {
|
let Some(job) = sup.job_mut(id) else {
|
||||||
drop(sup);
|
drop(sup);
|
||||||
return Ok(job_miss_error(ctx.supervisor.as_ref(), id));
|
return Ok(job_miss_error(ctx.supervisor.as_ref(), id));
|
||||||
};
|
};
|
||||||
@@ -518,6 +519,13 @@ fn handle_check(ctx: &RequestContext, args: &Value) -> Result<Value> {
|
|||||||
let buf = job.output_buf.lock();
|
let buf = job.output_buf.lock();
|
||||||
(buf.tail(), buf.total_written())
|
(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 tail_truncated = (tail.len() as u64) < total_written;
|
||||||
let mut result = json!({
|
let mut result = json!({
|
||||||
"status": job_status_str(status),
|
"status": job_status_str(status),
|
||||||
@@ -532,6 +540,11 @@ fn handle_check(ctx: &RequestContext, args: &Value) -> Result<Value> {
|
|||||||
result["message"] = json!(
|
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."
|
"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 {
|
} else {
|
||||||
result["message"] = json!(format!(
|
result["message"] = json!(format!(
|
||||||
"Job finished — retrieve the result with job__collect --id {id}"
|
"Job finished — retrieve the result with job__collect --id {id}"
|
||||||
@@ -1175,6 +1188,7 @@ mod tests {
|
|||||||
})),
|
})),
|
||||||
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1640,6 +1654,7 @@ mod tests {
|
|||||||
})),
|
})),
|
||||||
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
};
|
};
|
||||||
ctx.supervisor
|
ctx.supervisor
|
||||||
.as_ref()
|
.as_ref()
|
||||||
@@ -1754,6 +1769,86 @@ mod tests {
|
|||||||
assert!(ctx.supervisor.as_ref().unwrap().read().has_job("j1"));
|
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]
|
#[test]
|
||||||
fn handle_collect_applies_tail_lines() {
|
fn handle_collect_applies_tail_lines() {
|
||||||
run_async(async {
|
run_async(async {
|
||||||
@@ -1777,6 +1872,7 @@ mod tests {
|
|||||||
})),
|
})),
|
||||||
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
};
|
};
|
||||||
ctx.supervisor
|
ctx.supervisor
|
||||||
.as_ref()
|
.as_ref()
|
||||||
@@ -1815,6 +1911,7 @@ mod tests {
|
|||||||
})),
|
})),
|
||||||
output_buf,
|
output_buf,
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
};
|
};
|
||||||
ctx.supervisor
|
ctx.supervisor
|
||||||
.as_ref()
|
.as_ref()
|
||||||
@@ -1862,6 +1959,7 @@ mod tests {
|
|||||||
state: Arc::clone(&state),
|
state: Arc::clone(&state),
|
||||||
output_buf,
|
output_buf,
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
};
|
};
|
||||||
ctx.supervisor
|
ctx.supervisor
|
||||||
.as_ref()
|
.as_ref()
|
||||||
@@ -2115,6 +2213,7 @@ mod tests {
|
|||||||
state: Arc::clone(&state),
|
state: Arc::clone(&state),
|
||||||
output_buf,
|
output_buf,
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
};
|
};
|
||||||
ctx.supervisor
|
ctx.supervisor
|
||||||
.as_ref()
|
.as_ref()
|
||||||
|
|||||||
@@ -2409,6 +2409,19 @@ fn polyfill_cmd_name<T: AsRef<Path>>(cmd_name: &str, bin_dir: &[T]) -> String {
|
|||||||
cmd_name
|
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)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct ToolCallTracker {
|
pub struct ToolCallTracker {
|
||||||
last_calls: VecDeque<ToolCall>,
|
last_calls: VecDeque<ToolCall>,
|
||||||
@@ -2430,6 +2443,9 @@ impl ToolCallTracker {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn check_loop(&self, new_call: &ToolCall) -> Option<String> {
|
pub fn check_loop(&self, new_call: &ToolCall) -> Option<String> {
|
||||||
|
if is_loop_tracker_exempt(&new_call.name) {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
if self.last_calls.len() < self.max_repeats {
|
if self.last_calls.len() < self.max_repeats {
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
@@ -2496,6 +2512,9 @@ impl ToolCallTracker {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn record_call(&mut self, call: ToolCall) {
|
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 {
|
if self.last_calls.len() >= self.chain_len * self.max_repeats {
|
||||||
self.last_calls.pop_front();
|
self.last_calls.pop_front();
|
||||||
}
|
}
|
||||||
@@ -2663,6 +2682,7 @@ mod tests {
|
|||||||
})),
|
})),
|
||||||
output_buf: Arc::new(parking_lot::Mutex::new(jobs::RingBuf::default())),
|
output_buf: Arc::new(parking_lot::Mutex::new(jobs::RingBuf::default())),
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
};
|
};
|
||||||
let mut sup = crate::supervisor::Supervisor::new(0, 3).with_max_concurrent_jobs(4);
|
let mut sup = crate::supervisor::Supervisor::new(0, 3).with_max_concurrent_jobs(4);
|
||||||
sup.register(handle).unwrap();
|
sup.register(handle).unwrap();
|
||||||
@@ -3105,6 +3125,46 @@ mod tests {
|
|||||||
assert!(msg.contains("repeat_tool"));
|
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]
|
#[test]
|
||||||
fn prefix_constants_are_correct() {
|
fn prefix_constants_are_correct() {
|
||||||
assert_eq!(TODO_FUNCTION_PREFIX, "todo__");
|
assert_eq!(TODO_FUNCTION_PREFIX, "todo__");
|
||||||
|
|||||||
@@ -1635,6 +1635,7 @@ mod tests {
|
|||||||
})),
|
})),
|
||||||
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -67,8 +67,8 @@ pub struct JobHandle {
|
|||||||
pub abort_signal: AbortSignal,
|
pub abort_signal: AbortSignal,
|
||||||
pub state: Arc<Mutex<JobState>>,
|
pub state: Arc<Mutex<JobState>>,
|
||||||
pub output_buf: Arc<Mutex<RingBuf>>,
|
pub output_buf: Arc<Mutex<RingBuf>>,
|
||||||
#[allow(dead_code)]
|
|
||||||
pub no_change_checks: u32,
|
pub no_change_checks: u32,
|
||||||
|
pub last_check_state: Option<(JobStatus, u64)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl JobHandle {
|
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<Item = &JobHandle> {
|
pub fn jobs(&self) -> impl Iterator<Item = &JobHandle> {
|
||||||
self.handles.values().filter_map(|handle| match handle {
|
self.handles.values().filter_map(|handle| match handle {
|
||||||
TaskHandle::Job(handle) => Some(handle),
|
TaskHandle::Job(handle) => Some(handle),
|
||||||
@@ -392,6 +399,7 @@ mod tests {
|
|||||||
})),
|
})),
|
||||||
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
output_buf: Arc::new(Mutex::new(RingBuf::default())),
|
||||||
no_change_checks: 0,
|
no_change_checks: 0,
|
||||||
|
last_check_state: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user