From 56e355becf53ed94329b3970de92cd307c62cd76 Mon Sep 17 00:00:00 2001 From: Felix <141124084+Felyx-Fu@users.noreply.github.com> Date: Wed, 24 Jun 2026 02:53:49 +0800 Subject: [PATCH] fix(tools): preserve hidden follow-up handles (#4012) Co-authored-by: MackJack023 <141124084+MackJack023@users.noreply.github.com> --- .../agent_orchestration/running_subagents.rs | 59 +++++++++++++++++++ .../tools/wait_subagent.rs | 26 +++++++- src/openhuman/integrations/tools/apify.rs | 9 ++- .../integrations/tools/apify_tests.rs | 24 ++++++++ src/openhuman/search/tools/parallel.rs | 2 - src/openhuman/search/tools/parallel_tests.rs | 20 ++++++- src/openhuman/tools/ops_tests.rs | 7 +-- 7 files changed, 134 insertions(+), 13 deletions(-) diff --git a/src/openhuman/agent_orchestration/running_subagents.rs b/src/openhuman/agent_orchestration/running_subagents.rs index 5691cfc47..391a3b7d3 100644 --- a/src/openhuman/agent_orchestration/running_subagents.rs +++ b/src/openhuman/agent_orchestration/running_subagents.rs @@ -62,6 +62,13 @@ struct RunningSubagentEntry { status: watch::Receiver, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SubagentResumeRef { + pub task_id: String, + pub agent_id: String, + pub subagent_session_id: Option, +} + /// Soft cap on registry size. Terminal entries are only swept when the table /// grows past this, so the common case (a handful of live sub-agents) never /// evicts a still-uncollected terminal result out from under a `wait`/`steer`. @@ -159,6 +166,22 @@ pub fn task_id_for_session( Err(WaitError::Unknown) } +pub fn resume_ref_for_task( + task_id: &str, + parent_session: &str, +) -> Result { + let map = registry().lock().expect("running_subagents mutex poisoned"); + let entry = map.get(task_id).ok_or(WaitError::Unknown)?; + if entry.parent_session != parent_session { + return Err(WaitError::NotOwned); + } + Ok(SubagentResumeRef { + task_id: task_id.to_string(), + agent_id: entry.agent_id.clone(), + subagent_session_id: entry.subagent_session_id.clone(), + }) +} + /// Why a steer could not be delivered. #[derive(Debug, PartialEq, Eq)] pub enum SteerError { @@ -525,6 +548,42 @@ mod tests { prune("task-session"); } + #[tokio::test] + async fn resume_ref_for_task_includes_resume_fields_and_enforces_ownership() { + let _guard = test_guard(); + let (tx, rx) = status_channel(); + register( + "task-resume".into(), + "researcher".into(), + "session-owner".into(), + Some("subsess-resume".into()), + std::env::temp_dir(), + Some("thread-1".into()), + RunQueue::new(), + dummy_abort(), + rx, + ); + + let reference = + resume_ref_for_task("task-resume", "session-owner").expect("resume reference"); + assert_eq!(reference.task_id, "task-resume"); + assert_eq!(reference.agent_id, "researcher"); + assert_eq!( + reference.subagent_session_id.as_deref(), + Some("subsess-resume") + ); + assert!(matches!( + resume_ref_for_task("task-resume", "session-other"), + Err(WaitError::NotOwned) + )); + + let _ = tx.send(SubagentStatus::Completed { + output: "done".into(), + iterations: 1, + }); + prune("task-resume"); + } + #[tokio::test] async fn task_id_for_session_prefers_live_task_over_terminal_task() { let _guard = test_guard(); diff --git a/src/openhuman/agent_orchestration/tools/wait_subagent.rs b/src/openhuman/agent_orchestration/tools/wait_subagent.rs index 7140e644d..4054e1499 100644 --- a/src/openhuman/agent_orchestration/tools/wait_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/wait_subagent.rs @@ -135,6 +135,9 @@ impl Tool for WaitSubagentTool { timeout_secs ); + let resume_ref = + running_subagents::resume_ref_for_task(&resolved_task_id, &parent_session).ok(); + match running_subagents::wait( &resolved_task_id, &parent_session, @@ -158,10 +161,29 @@ impl Tool for WaitSubagentTool { resolved_task_id, question.chars().count() ); - Ok(ToolResult::success(format!( + let mut message = format!( "Sub-agent paused for clarification and did not finish: {question}\n\n\ It cannot proceed unattended. Resume it with continue_subagent once you have an answer." - ))) + ); + if let Some(reference) = resume_ref { + message.push_str("\n\n[subagent_resume_ref]\n"); + message.push_str( + &serde_json::to_string(&serde_json::json!({ + "task_id": reference.task_id, + "agent_id": reference.agent_id, + "subagent_session_id": reference.subagent_session_id, + "tool": "continue_subagent" + })) + .unwrap_or_else(|_| "{}".to_string()), + ); + message.push_str("\n[/subagent_resume_ref]"); + } else { + log::debug!( + "[wait_subagent] resume_ref_unavailable task_id={}", + resolved_task_id + ); + } + Ok(ToolResult::success(message)) } Ok(WaitOutcome::Terminal(SubagentStatus::Failed { error })) => { log::debug!( diff --git a/src/openhuman/integrations/tools/apify.rs b/src/openhuman/integrations/tools/apify.rs index 406f19f52..d8d4c1de2 100644 --- a/src/openhuman/integrations/tools/apify.rs +++ b/src/openhuman/integrations/tools/apify.rs @@ -64,8 +64,13 @@ fn format_run_actor_response(resp: &ApifyRunResponse, sync: bool) -> String { lines.push("Sample results:".to_string()); lines.push(summarize_json_array(items, 3)); } - } else if !sync { - lines.push("This run is still in progress. Poll with apify_get_run_status.".to_string()); + } else { + let follow_up = if sync { + "No inline result items were returned. Poll with apify_get_run_status or fetch results with apify_get_run_results." + } else { + "This run is still in progress. Poll with apify_get_run_status." + }; + lines.push(follow_up.to_string()); lines .push("Use the structured reference below for follow-up Apify tool calls.".to_string()); lines.push(format!( diff --git a/src/openhuman/integrations/tools/apify_tests.rs b/src/openhuman/integrations/tools/apify_tests.rs index 9c5092933..d7a17c149 100644 --- a/src/openhuman/integrations/tools/apify_tests.rs +++ b/src/openhuman/integrations/tools/apify_tests.rs @@ -148,6 +148,30 @@ fn run_actor_async_output_shows_polling_instruction() { assert!(output.contains("\"run_id\":\"run-456\"")); } +#[test] +fn run_actor_sync_without_items_shows_follow_up_ref() { + let resp = ApifyRunResponse { + run_id: "run-789".to_string(), + actor_id: "apify/crawler".to_string(), + status: "RUNNING".to_string(), + dataset_id: Some("dataset-789".to_string()), + items: None, + cost_usd: 0.05, + }; + + let output = format_run_actor_response(&resp, true); + let prose = output + .split("[apify_run_ref]") + .next() + .expect("prose before ref"); + + assert!(output.contains("Poll with apify_get_run_status")); + assert!(!prose.contains("run-789")); + assert!(output.contains("[apify_run_ref]")); + assert!(output.contains("\"run_id\":\"run-789\"")); + assert!(output.contains("\"dataset_id\":\"dataset-789\"")); +} + #[test] fn run_status_output_hides_internal_ids() { let resp = ApifyRunResponse { diff --git a/src/openhuman/search/tools/parallel.rs b/src/openhuman/search/tools/parallel.rs index 61bd9a7c4..44a7dd655 100644 --- a/src/openhuman/search/tools/parallel.rs +++ b/src/openhuman/search/tools/parallel.rs @@ -615,7 +615,6 @@ fn format_research_response(resp: ResearchResponse) -> Result { fn research_payload(resp: &ResearchResponse, display: &str) -> serde_json::Value { json!({ "display": display, - "run_id": resp.run_id, "status": resp.status, "result": resp.result, "cost_usd": resp.cost_usd, @@ -780,7 +779,6 @@ fn format_enrich_response(resp: EnrichResponse) -> Result { fn enrich_payload(resp: &EnrichResponse, display: &str) -> serde_json::Value { json!({ "display": display, - "run_id": resp.run_id, "status": resp.status, "output": resp.output, "cost_usd": resp.cost_usd, diff --git a/src/openhuman/search/tools/parallel_tests.rs b/src/openhuman/search/tools/parallel_tests.rs index b1993b340..eb70759f1 100644 --- a/src/openhuman/search/tools/parallel_tests.rs +++ b/src/openhuman/search/tools/parallel_tests.rs @@ -177,36 +177,52 @@ fn extract_response_with_full_content() { #[test] fn research_output_hides_internal_run_id() { - let output = format_research_response(ResearchResponse { + let resp = ResearchResponse { run_id: Some("run_internal_123".into()), status: Some("completed".into()), result: Some(json!({ "summary": "useful answer" })), cost_usd: 0.1234, + }; + let output = format_research_response(ResearchResponse { + run_id: resp.run_id.clone(), + status: resp.status.clone(), + result: resp.result.clone(), + cost_usd: resp.cost_usd, }) .unwrap(); + let payload = research_payload(&resp, &output); assert!(output.contains("Status: completed")); assert!(output.contains("useful answer")); assert!(output.contains("Cost: $0.1234")); assert!(!output.contains("Run:")); assert!(!output.contains("run_internal_123")); + assert!(payload.get("run_id").is_none()); } #[test] fn enrich_output_hides_internal_run_id() { - let output = format_enrich_response(EnrichResponse { + let resp = EnrichResponse { run_id: Some("run_internal_456".into()), status: Some("completed".into()), output: Some(json!({ "company": "OpenHuman" })), cost_usd: 0.5678, + }; + let output = format_enrich_response(EnrichResponse { + run_id: resp.run_id.clone(), + status: resp.status.clone(), + output: resp.output.clone(), + cost_usd: resp.cost_usd, }) .unwrap(); + let payload = enrich_payload(&resp, &output); assert!(output.contains("Status: completed")); assert!(output.contains("OpenHuman")); assert!(output.contains("Cost: $0.5678")); assert!(!output.contains("Run:")); assert!(!output.contains("run_internal_456")); + assert!(payload.get("run_id").is_none()); } #[test] diff --git a/src/openhuman/tools/ops_tests.rs b/src/openhuman/tools/ops_tests.rs index bc9baf1f5..5358608d6 100644 --- a/src/openhuman/tools/ops_tests.rs +++ b/src/openhuman/tools/ops_tests.rs @@ -1295,10 +1295,7 @@ async fn all_tools_executes_parallel_and_web_search_family_against_fake_backend( assert!(research_display.contains("\"company\": \"Tiny Humans\"")); assert!(!research_display.contains("research-core")); let research_payload = only_json_content(&research); - assert_eq!( - research_payload["run_id"], - serde_json::json!("research-core") - ); + assert!(research_payload.get("run_id").is_none()); let enrich = find_tool(&tools, "parallel_enrich") .execute(serde_json::json!({ @@ -1313,7 +1310,7 @@ async fn all_tools_executes_parallel_and_web_search_family_against_fake_backend( assert!(enrich_display.contains("\"inputEcho\": \"Tiny Humans\"")); assert!(!enrich_display.contains("enrich-1")); let enrich_payload = only_json_content(&enrich); - assert_eq!(enrich_payload["run_id"], serde_json::json!("enrich-1")); + assert!(enrich_payload.get("run_id").is_none()); let dataset = find_tool(&tools, "parallel_dataset") .execute(serde_json::json!({