feat: non-blocking background delegation for delegate_task - #7701
feat: non-blocking background delegation for delegate_task#7701murapadev wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
Adds a non-blocking (background=True) execution mode to delegate_task, persisting results to disk for later retrieval via session_id.
Changes:
- Introduces background result persistence (atomic write, retrieval, and cleanup) keyed by a validated 32-hex session ID.
- Refactors delegation execution into a shared
_execute_delegation(...)pipeline used by both blocking and background paths. - Adds a new test suite covering background behavior, session ID validation, atomic writes, and cleanup.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 8 comments.
| File | Description |
|---|---|
| tools/delegate_tool.py | Adds background execution mode, result persistence/retrieval utilities, session ID validation, and refactors core delegation pipeline. |
| tests/tools/test_delegate.py | Adds unit tests for background delegation behavior, session ID handling, atomic write behavior, and expired-result cleanup. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| # Reconstruct a lightweight agent-like object from the snapshot | ||
| class _IsolatedAgent: | ||
| """Stand-in for parent_agent in the background thread.""" | ||
| def __init__(self, state): | ||
| for k, v in state.items(): | ||
| setattr(self, k, v) | ||
|
|
||
| parent_agent = _IsolatedAgent(isolated_state) | ||
|
|
There was a problem hiding this comment.
_execute_delegation always reconstructs an _IsolatedAgent from isolated_state, which means even the blocking (non-background) path loses access to live parent attributes like valid_tool_names, _delegate_spinner, and _memory_manager. This can change toolset inheritance (falls back to DEFAULT_TOOLSETS when enabled_toolsets is None), disable progress display, and skip memory-provider callbacks. Consider passing the real parent_agent through for the synchronous path (and only isolating for background threads), or extend the isolated snapshot to include immutable copies of required fields (e.g., valid_tool_names) while keeping mutable objects out of the background path.
| # Reconstruct a lightweight agent-like object from the snapshot | |
| class _IsolatedAgent: | |
| """Stand-in for parent_agent in the background thread.""" | |
| def __init__(self, state): | |
| for k, v in state.items(): | |
| setattr(self, k, v) | |
| parent_agent = _IsolatedAgent(isolated_state) | |
| # Reconstruct a lightweight agent-like object only when we do not have a | |
| # live parent agent available. The sync path may rely on live attributes | |
| # (for example tool inheritance, progress UI, or memory callbacks) that | |
| # are intentionally not present in a background-safe snapshot. | |
| class _IsolatedAgent: | |
| """Stand-in for parent_agent when only isolated state is available.""" | |
| def __init__(self, state): | |
| for k, v in state.items(): | |
| setattr(self, k, v) | |
| live_parent_agent = None | |
| snapshot_state = isolated_state | |
| if isinstance(isolated_state, dict): | |
| # Allow callers to provide a live parent only for the synchronous path | |
| # while still using a plain serializable snapshot for background work. | |
| live_parent_agent = isolated_state.get("_live_parent_agent") | |
| snapshot_state = { | |
| k: v for k, v in isolated_state.items() | |
| if k != "_live_parent_agent" | |
| } | |
| if target_session_id is None and live_parent_agent is not None: | |
| parent_agent = live_parent_agent | |
| elif isinstance(snapshot_state, dict): | |
| parent_agent = _IsolatedAgent(snapshot_state) | |
| else: | |
| # Backward-compatible fallback: if a real agent-like object is passed | |
| # directly, preserve it instead of stripping live attributes. | |
| parent_agent = snapshot_state |
There was a problem hiding this comment.
Fixed. The sync path now passes directly (line 1101-1107 in delegate_task). Only the background path uses from the snapshot. The function accepts both and — it uses whichever is provided (not both).
| @@ -599,20 +794,31 @@ def delegate_task( | |||
| override_acp_command=t.get("acp_command") or acp_command, | |||
| override_acp_args=t.get("acp_args") or acp_args, | |||
| ) | |||
| # Override with correct parent tool names (before child construction mutated global) | |||
| child._delegate_saved_tool_names = _parent_tool_names | |||
| children.append((i, t, child)) | |||
| finally: | |||
| # Authoritative restore: reset global to parent's tool names after all children built | |||
| _model_tools._last_resolved_tool_names = _parent_tool_names | |||
There was a problem hiding this comment.
The background implementation still relies on the process-global model_tools._last_resolved_tool_names being temporarily mutated during child construction. With background threads, this becomes a cross-thread race: other tool calls (notably execute_code sandbox generation) can observe the child's tool list while this thread is between the save/restore points. To make this thread-safe, guard reads/writes of _last_resolved_tool_names with a shared lock, or refactor to avoid using a process-global for per-session tool resolution (e.g., thread-local or explicit propagation only).
There was a problem hiding this comment.
Fixed. Added threading.Lock around all reads/writes of _last_resolved_tool_names in model_tools.py (line ~198: _last_resolved_tool_names_lock) and in delegate_tool.py at lines ~517, ~656, ~838, ~841.
| if background: | ||
| bg_session_id = uuid.uuid4().hex # 32 hex characters | ||
| isolated_state = _isolate_agent_state(parent_agent) | ||
| stop_event = threading.Event() | ||
| thread = threading.Thread( | ||
| target=_run_in_background, | ||
| args=( | ||
| bg_session_id, goal, context, toolsets, tasks, | ||
| max_iterations, acp_command, acp_args, isolated_state, | ||
| stop_event, | ||
| ), |
There was a problem hiding this comment.
A stop_event is created and passed into the background thread, but nothing retains a reference to it (it’s not stored on the parent agent or in a registry keyed by session_id). As a result, callers have no way to trigger the documented graceful shutdown/cancellation behavior. If cancellation is a requirement, expose a way to signal stop (e.g., store the event by session_id and add a cancel API/tool action, or accept an externally-managed stop_event).
There was a problem hiding this comment.
Fixed. Added _BACKGROUND_STOP_EVENTS dict and _BACKGROUND_STOP_EVENTS_LOCK to store stop events keyed by session_id. Added cancel_background_delegation(session_id) public function that looks up the event and calls .set(). The _background_thread_main wrapper calls _discard_background_stop_event in finally block.
| # Atomic write: write to temp file in same partition, then rename | ||
| fd, tmp_path = tempfile.mkstemp(dir=results_dir, suffix='.json') | ||
| try: | ||
| with os.fdopen(fd, 'w') as f: |
There was a problem hiding this comment.
_save_background_result writes JSON using os.fdopen(fd, 'w') without specifying an encoding, but reads later use encoding='utf-8'. On systems where the default text encoding isn’t UTF-8, this can corrupt non-ASCII output or fail on read. Open the temp file with an explicit UTF-8 encoding to match the reader.
| with os.fdopen(fd, 'w') as f: | |
| with os.fdopen(fd, 'w', encoding='utf-8') as f: |
There was a problem hiding this comment.
Fixed. os.fdopen(fd, 'w', encoding='utf-8') in _save_background_result (line ~79).
| "description": ( | ||
| "Pass a session_id to retrieve results from a previously launched " | ||
| "background delegation. When provided without goal/tasks, returns the " | ||
| "saved result. session_id is the UUID prefix returned by background=True." |
There was a problem hiding this comment.
The schema description for session_id says it is a “UUID prefix”, but the implementation validates a full 32-character lowercase hex string (uuid4().hex). This mismatch is likely to confuse tool callers and breaks if they provide a shorter prefix. Update the description to match the enforced format.
| "saved result. session_id is the UUID prefix returned by background=True." | |
| "saved result. session_id must be the full 32-character lowercase " | |
| "hexadecimal ID returned by background=True." |
There was a problem hiding this comment.
Fixed. Schema description updated: "session_id must be the full 32-character lowercase hex string returned by background=True." (line ~1393 in DELEGATE_TASK_SCHEMA).
| original_bg = dt._run_in_background | ||
|
|
||
| def slow_bg(*args, **kwargs): | ||
| # Simulate a 5-second task but this should NOT block the caller | ||
| time.sleep(5) | ||
| original_bg(*args, **kwargs) | ||
|
|
There was a problem hiding this comment.
This test patches _run_in_background with slow_bg, but slow_bg eventually calls the real original_bg in a daemon thread. That background thread can run real delegation logic (AIAgent construction, tool execution, file writes) after the test completes, which can introduce flakiness and cross-test side effects. Prefer patching _run_in_background to a pure no-op or to a deterministic stub that only records inputs and/or writes a trivial result file.
There was a problem hiding this comment.
Fixed. test_background_returns_immediately now patches _background_thread_main (the actual thread target) with a pure no-op stub that only records inputs and calls time.sleep(0.05). No real delegation logic runs in the test.
| def test_immediate_return_under_100ms(self): | ||
| """Verify background=True returns within 100ms.""" | ||
| import tools.delegate_tool as dt | ||
|
|
||
| parent = _make_mock_parent(depth=0) | ||
|
|
||
| def instant_bg(*args, **kwargs): | ||
| # Complete instantly without sleeping | ||
| pass | ||
|
|
||
| with patch.object(dt, '_run_in_background', instant_bg): | ||
| start = time.monotonic() | ||
| result = json.loads(delegate_task(goal="Quick test", background=True, parent_agent=parent)) | ||
| elapsed_ms = (time.monotonic() - start) * 1000 | ||
|
|
||
| self.assertLess(elapsed_ms, 100, f"background=True took {elapsed_ms:.1f}ms to return") | ||
| self.assertTrue(result.get("background")) | ||
| self.assertEqual(len(result["session_id"]), 32) | ||
|
|
There was a problem hiding this comment.
The 100ms wall-clock threshold is very tight for thread creation/scheduling on loaded CI machines and is likely to be flaky. Consider relaxing the limit (e.g., a few hundred ms) or asserting non-blocking behavior via synchronization (e.g., ensure the call returns before the stubbed background function signals completion) rather than timing.
| def test_immediate_return_under_100ms(self): | |
| """Verify background=True returns within 100ms.""" | |
| import tools.delegate_tool as dt | |
| parent = _make_mock_parent(depth=0) | |
| def instant_bg(*args, **kwargs): | |
| # Complete instantly without sleeping | |
| pass | |
| with patch.object(dt, '_run_in_background', instant_bg): | |
| start = time.monotonic() | |
| result = json.loads(delegate_task(goal="Quick test", background=True, parent_agent=parent)) | |
| elapsed_ms = (time.monotonic() - start) * 1000 | |
| self.assertLess(elapsed_ms, 100, f"background=True took {elapsed_ms:.1f}ms to return") | |
| self.assertTrue(result.get("background")) | |
| self.assertEqual(len(result["session_id"]), 32) | |
| def test_background_returns_before_worker_completes(self): | |
| """Verify background=True returns before the background worker finishes.""" | |
| import tools.delegate_tool as dt | |
| parent = _make_mock_parent(depth=0) | |
| worker_started = threading.Event() | |
| allow_worker_to_finish = threading.Event() | |
| worker_finished = threading.Event() | |
| def blocking_bg(session_id, goal, context, toolsets, tasks, | |
| max_iterations, acp_command, acp_args, isolated_state, | |
| stop_event=None): | |
| worker_started.set() | |
| allow_worker_to_finish.wait(timeout=2.0) | |
| dt._save_background_result(session_id, { | |
| "results": [], | |
| "total_duration_seconds": 0.0, | |
| }) | |
| worker_finished.set() | |
| with patch.object(dt, '_run_in_background', blocking_bg): | |
| result = json.loads(delegate_task(goal="Quick test", background=True, parent_agent=parent)) | |
| returned_before_completion = not worker_finished.is_set() | |
| self.assertTrue(result.get("background")) | |
| self.assertEqual(len(result["session_id"]), 32) | |
| self.assertTrue(worker_started.wait(timeout=1.0), "background worker did not start") | |
| self.assertTrue(returned_before_completion, "delegate_task should return before background work completes") | |
| allow_worker_to_finish.set() | |
| self.assertTrue(worker_finished.wait(timeout=1.0), "background worker did not finish after being released") |
There was a problem hiding this comment.
Fixed. Replaced wall-clock timing with threading.Event-based synchronization. The test now asserts returned_before_completion = not worker_finished.is_set() after confirming worker_started.wait(timeout=1.0). No timing thresholds needed.
|
|
||
| def capture_isolated_state(session_id, goal, context, toolsets, tasks, | ||
| max_iterations, acp_command, acp_args, isolated_state, | ||
| stop_event=None): | ||
| # Modify the isolated state dict (should NOT affect parent) | ||
| isolated_state["_delegate_depth"] = 99 | ||
| isolated_state["some_attr"] = "modified_in_background" | ||
| captured_state["isolated_depth"] = isolated_state["_delegate_depth"] | ||
| captured_state["isolated_attr"] = isolated_state["some_attr"] | ||
| # Save a completed result so the background function completes | ||
| dt._save_background_result(session_id, { | ||
| "results": [], | ||
| "total_duration_seconds": 0.0, | ||
| }) | ||
|
|
||
| with patch.object(dt, '_run_in_background', capture_isolated_state): | ||
| result = json.loads(delegate_task(goal="Isolation test", background=True, parent_agent=parent)) | ||
| time.sleep(0.1) # Give the thread time to complete |
There was a problem hiding this comment.
This test uses time.sleep(0.1) to “give the thread time to complete” before asserting on captured_state. That can be flaky under CI load. Use a threading.Event (set from the patched _run_in_background) or poll for a condition (with a timeout) so the test deterministically waits for the background stub to run.
| def capture_isolated_state(session_id, goal, context, toolsets, tasks, | |
| max_iterations, acp_command, acp_args, isolated_state, | |
| stop_event=None): | |
| # Modify the isolated state dict (should NOT affect parent) | |
| isolated_state["_delegate_depth"] = 99 | |
| isolated_state["some_attr"] = "modified_in_background" | |
| captured_state["isolated_depth"] = isolated_state["_delegate_depth"] | |
| captured_state["isolated_attr"] = isolated_state["some_attr"] | |
| # Save a completed result so the background function completes | |
| dt._save_background_result(session_id, { | |
| "results": [], | |
| "total_duration_seconds": 0.0, | |
| }) | |
| with patch.object(dt, '_run_in_background', capture_isolated_state): | |
| result = json.loads(delegate_task(goal="Isolation test", background=True, parent_agent=parent)) | |
| time.sleep(0.1) # Give the thread time to complete | |
| completed_event = threading.Event() | |
| def capture_isolated_state(session_id, goal, context, toolsets, tasks, | |
| max_iterations, acp_command, acp_args, isolated_state, | |
| stop_event=None): | |
| try: | |
| # Modify the isolated state dict (should NOT affect parent) | |
| isolated_state["_delegate_depth"] = 99 | |
| isolated_state["some_attr"] = "modified_in_background" | |
| captured_state["isolated_depth"] = isolated_state["_delegate_depth"] | |
| captured_state["isolated_attr"] = isolated_state["some_attr"] | |
| # Save a completed result so the background function completes | |
| dt._save_background_result(session_id, { | |
| "results": [], | |
| "total_duration_seconds": 0.0, | |
| }) | |
| finally: | |
| completed_event.set() | |
| with patch.object(dt, '_run_in_background', capture_isolated_state): | |
| result = json.loads(delegate_task(goal="Isolation test", background=True, parent_agent=parent)) | |
| self.assertTrue( | |
| completed_event.wait(timeout=1.0), | |
| "Timed out waiting for background stub to complete", | |
| ) |
There was a problem hiding this comment.
Fixed. Replaced time.sleep(0.1) with completed_event.wait(timeout=1.0). The patched function sets completed_event in a finally block.
9253712 to
065ac65
Compare
- Add background=True param to delegate_task: launches subagent in a daemon thread and returns immediately with a session_id (UUID prefix) - Add session_id param: retrieve results from a previous background delegation by passing the session_id (no goal/tasks needed) - Results stored in ~/.hermes/delegation_results/<session_id>.json - Background thread saves result atomically on completion - Thread-safe: _last_resolved_tool_names restored in finally block - Schema updated with NON-BLOCKING MODE section and both new params This enables parent agent to stay responsive while subagents run, supporting real concurrent workflows (e.g. research + coding in parallel while continuing to chat with user).
065ac65 to
52ffe1b
Compare
alt-glitch
added
type/feature
|
Related to #8482 (competing async_delegation toolset implementation). Both address non-blocking background delegation but with different approaches. |
Merge origin/main into feat/non-blocking-delegate: - Keep non-blocking background mode (background, session_id params) - Keep upstream spawn pause kill switch - Keep upstream configurable depth (max_spawn_depth) - Keep upstream configurable concurrency (max_concurrent_children) - Keep upstream role parameter (leaf/orchestrator) - Combine both in tool description and registry handler
|
Thanks for the non-blocking delegation work. Current
Automated hermes-sweeper review. |
teknium1
added
the
sweeper:implemented-on-main
Context
Replaces #6813 (closed). All three issues raised by teknium1 have been addressed in this revision.
What changed since #6813
1. Single entry point — NO pipeline duplication
The delegation logic is now in
_execute_delegation(...). Both sync and background paths call the same function. No duplicated code.2. Graceful shutdown via stop_event
Background thread accepts
threading.Eventand checks it at 5 logical iteration boundaries:If triggered: saves
{"error": "stopped_by_parent", "partial_results": [...]}and exits.3. Result file cleanup
_load_background_result)cleanup_expired_results(max_age_hours=24)prunes stale files4. State isolation
_isolate_agent_state(agent)snapshots all parent state into a plain dict. Background thread receivesisolated_statedict (not a live reference)._IsolatedAgentstand-in prevents any shared mutable state.5. Session ID: 32 hex chars, validated
^[a-f0-9]{32}$validated via_validate_session_id()before any file operations. Path traversal blocked.6. Atomic writes
tempfile.mkstemp+os.fdopen+flush+os.replace— write-then-move pattern.Tests
10 new tests added, 77 total passing (72 delegate + 5 toolset scope).
Changes summary
tools/delegate_tool.py: +526/-182 linestests/tools/test_delegate.py: +315 linesCommits:
64c206ef— feat: non-blocking background delegation37c53632— fix: correct delegation_results path in docstringCloses #6813