diff --git a/src/config/agent.rs b/src/config/agent.rs index 8ff23d8..2be8b31 100644 --- a/src/config/agent.rs +++ b/src/config/agent.rs @@ -1515,4 +1515,51 @@ nodes: {} assert_eq!(config.top_k, Some(7)); assert_eq!(config.embedding_model.as_deref(), Some("some:model")); } + + #[test] + fn interpolated_instructions_without_job_declarations_is_byte_identical_across_job_settings() { + let agent = |max_concurrent_jobs| { + Agent::test_new(AgentConfig { + instructions: "hi".to_string(), + max_concurrent_jobs, + ..AgentConfig::default() + }) + }; + + let baseline = agent(None).interpolated_instructions(); + assert!( + !baseline.contains(DEFAULT_JOB_INSTRUCTIONS), + "no job guidance may be injected without job__ declarations" + ); + assert_eq!(baseline, agent(Some(0)).interpolated_instructions()); + assert_eq!(baseline, agent(Some(7)).interpolated_instructions()); + + let mut with_unrelated = agent(None); + with_unrelated.functions.append_todo_functions(); + assert_eq!( + baseline, + with_unrelated.interpolated_instructions(), + "job guidance injection must key strictly on the job__ prefix" + ); + } + + #[test] + fn interpolated_instructions_with_job_declarations_appends_job_guidance() { + let config = AgentConfig { + instructions: "hi".to_string(), + ..AgentConfig::default() + }; + let baseline = Agent::test_new(config.clone()).interpolated_instructions(); + + let mut agent = Agent::test_new(config); + agent.functions.append_job_functions(); + let output = agent.interpolated_instructions(); + + assert!(output.contains(DEFAULT_JOB_INSTRUCTIONS)); + let expected = format!( + "hi\n{DEFAULT_JOB_INSTRUCTIONS}{}", + baseline.strip_prefix("hi").unwrap() + ); + assert_eq!(output, expected); + } } diff --git a/src/config/request_context.rs b/src/config/request_context.rs index 431177f..25f9204 100644 --- a/src/config/request_context.rs +++ b/src/config/request_context.rs @@ -7555,4 +7555,254 @@ mod tests { "global config" ); } + + #[test] + fn select_functions_preserves_job_tools_under_empty_role_filter() { + let mut ctx = create_test_ctx(); + ctx.tool_scope.functions.append_job_functions(); + + let mut role = Role::new("r", "p"); + role.set_enabled_tools(Some(vec![])); + + let fns = ctx.select_functions(&role).unwrap(); + let names: Vec<&str> = fns.iter().map(|f| f.name.as_str()).collect(); + assert_eq!( + names, + vec![ + "job__start", + "job__check", + "job__collect", + "job__cancel", + "job__list" + ], + "job__ tools must survive an empty role tool filter" + ); + } + + #[test] + #[serial] + fn select_functions_preserves_job_tools_under_empty_agent_filter() { + let _guard = TestConfigDirGuard::new(); + let mut ctx = create_test_ctx(); + let app = ctx.app.config.clone(); + let agent_name = format!( + "test_job_agent_{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let agent_dir = paths::agent_data_dir(&agent_name); + create_dir_all(&agent_dir).unwrap(); + write( + agent_dir.join("config.yaml"), + format!("name: {agent_name}\ninstructions: hi\n"), + ) + .unwrap(); + + let abort = utils::create_abort_signal(); + run_async(ctx.use_agent(&app, &agent_name, None, abort)).unwrap(); + + let mut role = Role::new("r", "p"); + role.set_enabled_tools(Some(vec![])); + + let fns = ctx.select_functions(&role).unwrap(); + let names: Vec<&str> = fns.iter().map(|f| f.name.as_str()).collect(); + assert!( + names.contains(&"job__start"), + "job__ tools must survive an empty agent tool filter, got: {names:?}" + ); + assert!(names.contains(&"job__collect")); + } + + #[test] + #[serial] + fn select_functions_when_jobs_disabled_is_byte_identical_to_no_jobs_baseline() { + let _guard = TestConfigDirGuard::new(); + let app_state = app_state_with_mcp_config(false, &[]); + let mut ctx = RequestContext::new(app_state, WorkingMode::Repl); + let app = ctx.app.config.clone(); + let abort = utils::create_abort_signal(); + + let mut role = Role::new("r", "p"); + role.set_enabled_tools(Some(vec!["all".to_string()])); + + let jobs_off = AppConfig { + max_concurrent_jobs: Some(0), + ..(*app).clone() + }; + run_async(ctx.rebuild_tool_scope(&jobs_off, None, abort.clone())).unwrap(); + let without_jobs = serde_json::to_string(&ctx.select_functions(&role)).unwrap(); + assert!( + !without_jobs.contains("job__"), + "no job__ declarations may leak when jobs are disabled, got: {without_jobs}" + ); + + run_async(ctx.rebuild_tool_scope(&app, None, abort)).unwrap(); + let with_jobs = ctx.select_functions(&role).unwrap(); + assert!(with_jobs.iter().any(|f| f.name.starts_with("job__"))); + let stripped: Vec = with_jobs + .into_iter() + .filter(|f| !f.name.starts_with("job__")) + .collect(); + + assert_eq!( + without_jobs, + serde_json::to_string(&Some(stripped)).unwrap(), + "jobs-disabled tool list must be byte-identical to the jobs-enabled list minus job__ declarations" + ); + } + + #[test] + fn select_functions_returns_none_when_no_tools_enabled_and_jobs_disabled() { + let app_state = { + let config = AppConfig { + max_concurrent_jobs: Some(0), + ..AppConfig::default() + }; + Arc::new(AppState { + config: Arc::new(config), + vault: Arc::new(Vault::default()), + mcp_factory: Arc::new(McpFactory::default()), + rag_cache: Arc::new(RagCache::default()), + mcp_config: None, + mcp_log_path: None, + mcp_registry: None, + functions: Functions::default(), + }) + }; + let ctx = RequestContext::new(app_state, WorkingMode::Cmd); + assert!(ctx.select_functions(&Role::default()).is_none()); + } + + #[test] + fn tools_info_lists_job_tools_when_enabled() { + let mut ctx = create_test_ctx(); + ctx.tool_scope.functions.append_job_functions(); + + let info = ctx.tools_info().unwrap(); + + for name in [ + "job__start", + "job__check", + "job__collect", + "job__cancel", + "job__list", + ] { + assert!( + info.contains(name), + "expected {name} in output, got: {info}" + ); + } + } + + fn make_running_job(abort_signal: utils::AbortSignal) -> crate::supervisor::JobHandle { + // Leak the runtime so the spawned task is never polled and the job + // stays running for the duration of the test. + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let join_handle = rt.spawn(async { + Ok(crate::supervisor::JobResult { + output: serde_json::Value::Null, + exit_code: Some(0), + output_bytes_captured: 0, + }) + }); + std::mem::forget(rt); + crate::supervisor::JobHandle { + id: "j1".to_string(), + tool: "execute_command".to_string(), + started_at: std::time::Instant::now(), + join_handle, + abort_signal, + state: Arc::new(parking_lot::Mutex::new(crate::supervisor::JobState { + status: crate::supervisor::JobStatus::Running, + pgid: None, + })), + output_buf: Arc::new(parking_lot::Mutex::new( + crate::function::jobs::RingBuf::default(), + )), + no_change_checks: 0, + last_check_state: None, + } + } + + #[test] + #[serial] + fn use_agent_cancels_running_jobs_of_previous_supervisor() { + let _guard = TestConfigDirGuard::new(); + let mut ctx = create_test_ctx(); + let app = ctx.app.config.clone(); + let agent_name = format!( + "test_agent_{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let agent_dir = paths::agent_data_dir(&agent_name); + create_dir_all(&agent_dir).unwrap(); + write( + agent_dir.join("config.yaml"), + format!("name: {agent_name}\ninstructions: hi\n"), + ) + .unwrap(); + + let job_sig = utils::create_abort_signal(); + let old_sup = Arc::new(RwLock::new( + Supervisor::new(4, 3).with_max_concurrent_jobs(1), + )); + old_sup + .write() + .register(make_running_job(job_sig.clone())) + .unwrap(); + ctx.supervisor = Some(old_sup); + + run_async(ctx.use_agent(&app, &agent_name, None, utils::create_abort_signal())).unwrap(); + + assert!( + job_sig.aborted(), + "running jobs of the previous supervisor must be cancelled" + ); + assert!(ctx.supervisor.is_some()); + } + + #[test] + #[serial] + fn exit_agent_cancels_running_jobs() { + let _guard = TestConfigDirGuard::new(); + let mut ctx = create_test_ctx(); + let app = ctx.app.config.clone(); + + let job_sig = utils::create_abort_signal(); + let sup = Arc::new(RwLock::new( + Supervisor::new(4, 3).with_max_concurrent_jobs(1), + )); + sup.write() + .register(make_running_job(job_sig.clone())) + .unwrap(); + ctx.agent = Some(Agent::test_new(AgentConfig::default())); + ctx.supervisor = Some(sup); + + ctx.exit_agent(&app).unwrap(); + + assert!(job_sig.aborted(), "exit_agent must cancel running jobs"); + assert!(ctx.supervisor.is_none()); + } + + #[test] + fn toggle_tool_rejects_job_tools_as_unknown() { + let mut ctx = create_test_ctx(); + ctx.tool_scope.functions.append_job_functions(); + + for action in ["enable", "disable"] { + let err = ctx.toggle_tool(action, "job__start").unwrap_err(); + assert!( + err.to_string().contains("Unknown tool 'job__start'"), + "expected job__start to be rejected on {action}, got: {err}" + ); + } + } } diff --git a/src/function/jobs.rs b/src/function/jobs.rs index f2d6c92..7d140a4 100644 --- a/src/function/jobs.rs +++ b/src/function/jobs.rs @@ -1928,6 +1928,49 @@ mod tests { }); } + /// The completion notification is pushed from inside the job task, after + /// the run — a panic unwinds past the push, and neither the supervisor nor + /// collect synthesizes a notification for a panicked job. + #[test] + fn panicked_job_task_skips_the_completion_notification() { + run_async(async { + let ctx = ctx_with_job_supervisor(4); + let join_handle: tokio::task::JoinHandle> = + tokio::spawn(async { panic!("boom") }); + let handle = JobHandle { + id: "j1".to_string(), + tool: "execute_command".to_string(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state: Arc::new(Mutex::new(JobState { + // A panic never reaches the status update — the cell + // stays Running, which is the real post-panic state. + status: JobStatus::Running, + pgid: None, + })), + output_buf: Arc::new(Mutex::new(RingBuf::default())), + no_change_checks: 0, + last_check_state: None, + }; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(handle) + .unwrap(); + + let result = handle_collect(&ctx, &json!({"id": "j1"})).await.unwrap(); + + assert_eq!(result["status"], "failed"); + assert!(result["error"].as_str().unwrap().contains("panicked")); + assert!( + ctx.notification_queue.drain().is_empty(), + "a panicked job must never produce a completion notification" + ); + }); + } + #[cfg(unix)] #[test] fn handle_cancel_kills_running_process_job() { @@ -2323,4 +2366,301 @@ mod tests { .unwrap(); assert_eq!(tasks["tasks"].as_array().unwrap().len(), 1); } + + #[cfg(unix)] + #[test] + fn handle_cancel_kills_grandchild_process() { + run_async(async { + let ctx = ctx_with_job_supervisor(4); + let state = Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + let snapshot = test_snapshot("sh", &["-c", "sleep 30 & echo CHILD:$!; wait"], 0); + let task_state = Arc::clone(&state); + let task_buf = Arc::clone(&output_buf); + let join_handle = + tokio::spawn(async move { run_process_job(snapshot, task_state, task_buf).await }); + + let deadline = time::Instant::now() + Duration::from_secs(5); + let grandchild_pid = loop { + let tail = String::from_utf8_lossy(&output_buf.lock().tail()).to_string(); + if let Some(rest) = tail.split("CHILD:").nth(1) + && let Some(line_end) = rest.find('\n') + { + break rest[..line_end].trim().parse::().unwrap(); + } + assert!( + time::Instant::now() < deadline, + "grandchild pid never appeared in the ring buffer" + ); + time::sleep(Duration::from_millis(10)).await; + }; + + let handle = JobHandle { + id: "j1".to_string(), + tool: "sh".to_string(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state: Arc::clone(&state), + output_buf, + no_change_checks: 0, + last_check_state: None, + }; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(handle) + .unwrap(); + + let result = handle_cancel(&ctx, &json!({"id": "j1"})).await.unwrap(); + + assert_eq!(result["status"], "cancelled"); + // kill(pid, 0) alone can't observe the death: the orphaned + // grandchild lingers as an unreaped zombie under init/launchd, + // so a Z state also proves the group kill landed. + fn grandchild_is_dead(pid: i32) -> bool { + let esrch = unsafe { libc::kill(pid, 0) } == -1 + && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH); + if esrch { + return true; + } + let stat = std::process::Command::new("ps") + .args(["-o", "stat=", "-p", &pid.to_string()]) + .output() + .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_string()) + .unwrap_or_default(); + stat.is_empty() || stat.starts_with('Z') + } + let deadline = time::Instant::now() + Duration::from_secs(5); + while !grandchild_is_dead(grandchild_pid) { + assert!( + time::Instant::now() < deadline, + "grandchild must die with the process group" + ); + time::sleep(Duration::from_millis(10)).await; + } + }); + } + + #[cfg(unix)] + #[test] + fn run_process_job_clears_pgid_after_normal_completion() { + run_async(async { + let state = Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })); + let output_buf = Arc::new(Mutex::new(RingBuf::default())); + let snapshot = test_snapshot("echo", &["done"], 0); + + let result = run_process_job(snapshot, Arc::clone(&state), output_buf) + .await + .unwrap(); + + assert_eq!(result.exit_code, Some(0)); + assert!( + state.lock().pgid.is_none(), + "pid-reuse guard must clear pgid" + ); + }); + } + + #[test] + fn handle_start_rejects_shell_and_path_shaped_names_without_spawn() { + let mut ctx = plain_ctx(); + + for tool in ["bash", "./script.sh", "/usr/bin/env", "ls"] { + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": tool, "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("not enabled in this context") + ); + } + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + } + + #[cfg(unix)] + #[test] + fn handle_start_rejects_context_filtered_tool_and_accepts_in_filter() { + run_async(async { + let mut ctx = plain_ctx(); + ctx.declared_function_names.insert("echo".into()); + + let rejected = handle_start(&mut ctx, &json!({"tool": "git_command", "arguments": {}})) + .await + .unwrap(); + + assert_eq!(rejected["status"], "error"); + assert!( + rejected["message"] + .as_str() + .unwrap() + .contains("not enabled in this context") + ); + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + + let started = handle_start(&mut ctx, &json!({"tool": "echo", "arguments": {}})) + .await + .unwrap(); + + assert_eq!(started["status"], "ok"); + let job_id = started["job_id"].as_str().unwrap().to_string(); + let collected = handle_collect(&ctx, &json!({"id": job_id})).await.unwrap(); + assert_eq!(collected["status"], "completed"); + }); + } + + #[test] + fn handle_start_rejects_undeclared_mcp_invoke_without_spawn() { + let mut ctx = plain_ctx(); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "mcp_invoke_someserver", "arguments": {"tool": "search"}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("not enabled in this context") + ); + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + } + + #[test] + fn handle_start_rejects_declared_but_non_whitelisted_tools_without_spawn() { + let mut ctx = plain_ctx(); + for tool in ["memory__write", "fs_read", "agent__spawn", "user__select"] { + ctx.declared_function_names.insert(tool.into()); + } + + for (tool, category) in [ + ("memory__write", "mutates agent/session state"), + ("fs_read", "is fast"), + ("agent__spawn", "already asynchronous"), + ("user__select", "interactive"), + ] { + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": tool, "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!(result["message"].as_str().unwrap().contains(category)); + } + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + } + + #[test] + fn handle_start_rejects_mapping_tools_alias() { + let app_state = app_state_with_config(|config| { + config + .mapping_tools + .insert("shell".into(), "execute_command".into()); + }); + let mut ctx = RequestContext::new(app_state, WorkingMode::Cmd); + ctx.declared_function_names.insert("execute_command".into()); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "shell", "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("not enabled in this context") + ); + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + } + + #[test] + fn child_context_cannot_reach_parent_job_ids() { + run_async(async { + let parent = ctx_with_job_supervisor(4); + parent + .supervisor + .as_ref() + .unwrap() + .write() + .register(make_running_job("job_p1")) + .unwrap(); + let child = RequestContext::new_for_child( + default_app_state(), + &parent, + 1, + Arc::new(Inbox::new()), + "c1".into(), + ); + assert!(child.supervisor.is_none()); + + let checked = handle_check(&child, &json!({"id": "job_p1"})).unwrap(); + let collected = handle_collect(&child, &json!({"id": "job_p1"})) + .await + .unwrap(); + let cancelled = handle_cancel(&child, &json!({"id": "job_p1"})) + .await + .unwrap(); + + for result in [checked, collected, cancelled] { + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("No job 'job_p1' is registered") + ); + } + assert!(parent.supervisor.as_ref().unwrap().read().has_job("job_p1")); + }); + } + + #[test] + fn handle_start_ignores_mid_batch_tool_scope_additions() { + let mut ctx = plain_ctx(); + ctx.declared_function_names.insert("job__start".into()); + ctx.tool_scope + .functions + .declarations + .push(FunctionDeclaration { + name: "late_external_tool".into(), + description: String::new(), + parameters: JsonSchema::default(), + agent: false, + }); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "late_external_tool", "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!( + result["message"] + .as_str() + .unwrap() + .contains("not enabled in this context") + ); + assert!(ctx.supervisor.is_none(), "no job may be spawned"); + } } diff --git a/src/function/mod.rs b/src/function/mod.rs index e116dd0..7a96ec4 100644 --- a/src/function/mod.rs +++ b/src/function/mod.rs @@ -2940,6 +2940,51 @@ mod tests { ); } + #[cfg(unix)] + #[test] + fn job_finished_earlier_drains_notification_on_later_batch() { + run_async(async { + let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd); + ctx.declared_function_names.insert("echo".into()); + + let started = jobs::handle_job_tool( + &mut ctx, + "job__start", + &json!({"tool": "echo", "arguments": {}}), + ) + .await + .unwrap(); + assert_eq!(started["status"], "ok"); + let job_id = started["job_id"].as_str().unwrap().to_string(); + + let supervisor = ctx.supervisor.clone().unwrap(); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); + while tokio::time::Instant::now() < deadline { + let finished = supervisor + .read() + .job(&job_id) + .is_none_or(|job| job.join_handle.is_finished()); + if finished { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + + let calls = vec![call("unknown_tool", Some("id-2"))]; + let results = eval_tool_calls(&mut ctx, calls).await.unwrap(); + + let out = &results.last().unwrap().output; + assert_eq!(out["system_notifications"][0]["id"], job_id); + assert_eq!(out["system_notifications"][0]["event"], "job_completed"); + assert!( + out["notification_instruction"] + .as_str() + .unwrap() + .contains("next_action") + ); + }); + } + #[test] fn normalize_tool_result_preserves_non_null_values() { assert_eq!( @@ -3125,6 +3170,22 @@ mod tests { assert!(msg.contains("repeat_tool")); } + #[test] + fn loop_tracker_exempt_list_is_exactly_the_polling_tools() { + let actual: std::collections::HashSet<&str> = + LOOP_TRACKER_EXEMPT_TOOLS.iter().copied().collect(); + let expected: std::collections::HashSet<&str> = [ + "job__check", + "job__list", + "agent__check", + "agent__list_running", + ] + .into_iter() + .collect(); + assert_eq!(LOOP_TRACKER_EXEMPT_TOOLS.len(), 4); + assert_eq!(actual, expected); + } + #[test] fn tracker_exempt_tools_never_trip() { for name in LOOP_TRACKER_EXEMPT_TOOLS { diff --git a/src/function/supervisor.rs b/src/function/supervisor.rs index e1586ab..e276c47 100644 --- a/src/function/supervisor.rs +++ b/src/function/supervisor.rs @@ -2911,4 +2911,86 @@ mod tests { assert_job_teaching_error(&result, "bg_p"); }); } + + #[test] + fn guardrail_burns_bounded_injects_then_force_terminates_running_job() { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + + rt.block_on(async { + let mut ctx = ctx_with_job_capable_supervisor(); + let abort = create_abort_signal(); + let join_handle = tokio::spawn(async { + time::sleep(Duration::from_secs(60)).await; + Ok(JobResult { + output: json!(null), + exit_code: Some(0), + output_bytes_captured: 0, + }) + }); + let handle = JobHandle { + id: "job_1".to_string(), + tool: "execute_command".to_string(), + started_at: std::time::Instant::now(), + join_handle, + abort_signal: abort.clone(), + state: Arc::new(Mutex::new(JobState { + status: JobStatus::Running, + pgid: None, + })), + output_buf: Arc::new(Mutex::new(RingBuf::default())), + no_change_checks: 0, + last_check_state: None, + }; + ctx.supervisor + .as_ref() + .unwrap() + .write() + .register(handle) + .unwrap(); + + for expected_count in 1..=PENDING_AGENTS_GUARDRAIL_MAX { + match check_pending_agents_guardrail(&mut ctx) { + GuardrailAction::Inject(prompt) => assert!(prompt.contains("job_1")), + _ => panic!("expected Inject below max"), + } + assert_eq!(ctx.pending_agents_guardrail_count, expected_count); + } + + match check_pending_agents_guardrail(&mut ctx) { + GuardrailAction::ForceTerminate(ids) => { + assert_eq!(ids, vec!["job_1".to_string()]); + } + _ => panic!("expected ForceTerminate at max"), + } + assert_eq!(ctx.pending_agents_guardrail_count, 0); + assert!(abort.aborted()); + }); + } + + #[test] + fn guardrail_force_terminate_discards_finished_uncollected_job() { + let mut ctx = ctx_with_job_capable_supervisor(); + register_fake_job(&mut ctx, "job_1"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while !pending_tasks(&ctx).iter().any(|t| t.finished) { + assert!( + std::time::Instant::now() < deadline, + "job 'job_1' never finished" + ); + std::thread::sleep(Duration::from_millis(10)); + } + ctx.pending_agents_guardrail_count = PENDING_AGENTS_GUARDRAIL_MAX; + + match check_pending_agents_guardrail(&mut ctx) { + GuardrailAction::ForceTerminate(ids) => { + assert_eq!(ids, vec!["job_1".to_string()]); + } + _ => panic!("expected ForceTerminate action"), + } + assert_eq!(ctx.pending_agents_guardrail_count, 0); + assert!(!ctx.supervisor.as_ref().unwrap().read().has_job("job_1")); + } } diff --git a/src/graph/executor.rs b/src/graph/executor.rs index a53441d..70ec480 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -856,4 +856,86 @@ nodes: ); assert!(err.contains("sleeper"), "error should name frontier: {err}"); } + + #[cfg(unix)] + #[tokio::test] + async fn background_job_survives_graph_node_execution() { + if !cmd_available("bash") { + eprintln!("skipping: bash not available"); + return; + } + let ws = TestWorkspace::new(); + ws.write_script("noop.sh", "#!/bin/bash\necho '{}'\n"); + + let yaml = r#" +name: background_job_survival_test +start: noop +nodes: + noop: + type: script + script: noop.sh + state_updates: {} + next: done + done: + type: end + output: "done" +"#; + let graph: Graph = serde_yaml::from_str(yaml).unwrap(); + + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let join_handle = rt.spawn(async { + Ok(crate::supervisor::JobResult { + output: Value::Null, + exit_code: Some(0), + output_bytes_captured: 0, + }) + }); + std::mem::forget(rt); + let handle = crate::supervisor::JobHandle { + id: "job_bg".to_string(), + tool: "execute_command".to_string(), + started_at: Instant::now(), + join_handle, + abort_signal: create_abort_signal(), + state: Arc::new(parking_lot::Mutex::new(crate::supervisor::JobState { + status: crate::supervisor::JobStatus::Completed, + pgid: None, + })), + output_buf: Arc::new(parking_lot::Mutex::new( + crate::function::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(); + + let mut ctx = make_ctx(); + ctx.supervisor = Some(Arc::new(parking_lot::RwLock::new(sup))); + ctx.notification_queue + .push(crate::supervisor::notification::job_notification( + "job_bg", + "execute_command", + true, + )); + + let abort = create_abort_signal(); + let result = GraphExecutor::new(graph, &ws.dir) + .execute(&mut ctx, abort) + .await + .unwrap_or_else(|e| panic!("executor failed: {e:#}")); + assert_eq!(result, "done"); + + assert!( + ctx.supervisor.as_ref().unwrap().read().has_job("job_bg"), + "graph execution must not touch registered job handles" + ); + let events = ctx.notification_queue.drain(); + assert_eq!(events.len(), 1, "queued notification must survive the run"); + assert_eq!(events[0].id, "job_bg"); + assert_eq!(events[0].event, "job_completed"); + } }