diff --git a/tests/tools/test_async_delegation_stale_profile_scope.py b/tests/tools/test_async_delegation_stale_profile_scope.py new file mode 100644 index 0000000000..970326a8ac --- /dev/null +++ b/tests/tools/test_async_delegation_stale_profile_scope.py @@ -0,0 +1,112 @@ +"""Profile-scope regression for async-delegation stale finalization.""" + +import sqlite3 +import threading +import time + +import pytest + +from hermes_constants import reset_hermes_home_override, set_hermes_home_override +from tools import async_delegation as ad +from tools.process_registry import process_registry + + +@pytest.fixture(autouse=True) +def _clean_async_delegation_state(): + ad._reset_for_tests() + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + yield + ad._monitor_stop.set() + ad._reset_for_tests() + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + + +def _delegation_state(db_path, delegation_id): + if not db_path.exists(): + return None + conn = sqlite3.connect(db_path) + try: + row = conn.execute( + "SELECT state FROM async_delegations WHERE delegation_id=?", + (delegation_id,), + ).fetchone() + return None if row is None else row[0] + finally: + conn.close() + + +def test_stale_monitor_force_finalize_updates_origin_profile_ledger(tmp_path, monkeypatch): + """A bare monitor thread must settle the profile that dispatched the unit. + + Normal async workers carry the dispatching ContextVars via + ``propagate_context_to_thread``. The single stale-monitor thread does not: + Python starts it with an empty Context, so a forced finalization must not + re-resolve ``get_hermes_home()`` to the process launch profile. + """ + launch_home = tmp_path / "launch" + profile_home = launch_home / "profiles" / "secondary" + profile_home.mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(launch_home)) + + # We drive the production monitor loop ourselves below so the test controls + # exactly when the record becomes expired. + monkeypatch.setattr(ad, "_ensure_stale_monitor", lambda: None) + monkeypatch.setattr(ad, "_STALE_CHECK_INTERVAL", 0.01) + monkeypatch.setattr(ad, "_STALL_GRACE_SECONDS", 0.0) + + gate = threading.Event() + + def runner(): + gate.wait(timeout=30) + return {"status": "completed", "summary": "late runner result"} + + token = set_hermes_home_override(profile_home) + try: + handle = ad.dispatch_async_delegation( + goal="profile-owned stalled task", + context=None, + toolsets=None, + role="leaf", + model="test-model", + session_key="secondary-session", + runner=runner, + max_async_children=1, + progress_fn=lambda: ("frozen", False), + ) + finally: + reset_hermes_home_override(token) + + delegation_id = handle["delegation_id"] + profile_db = profile_home / "state.db" + launch_db = launch_home / "state.db" + assert _delegation_state(profile_db, delegation_id) == "running" + + # Put the live record exactly at the monitor's force-finalize boundary. + with ad._records_lock: + record = ad._records[delegation_id] + record.update( + status="stalling", + _started=True, + _interrupted_at=time.time() - 1, + _stall_quiet_seconds=1.0, + _stall_threshold_seconds=0.5, + _stall_in_tool=False, + ) + + # This is the production ownership shape: one daemon monitor thread serves + # every profile and therefore starts without the dispatching ContextVars. + ad._monitor_stop.clear() + monitor = threading.Thread(target=ad._stale_monitor_loop, daemon=True) + monitor.start() + monitor.join(timeout=2) + assert not monitor.is_alive() + + try: + assert _delegation_state(profile_db, delegation_id) == "stalled" + assert _delegation_state(launch_db, delegation_id) is None + finally: + # The real runner can return after the forced stall; _finalize must then + # be a no-op for the already-terminal record. + gate.set() diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 4659245120..99a0c83b89 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -555,6 +555,9 @@ def _dispatch( "slot_key": slot_key or delegation_id, # Which of the call's ``goals`` this unit runs (None = all of them). **({"task_indexes": list(task_indexes)} if task_indexes is not None else {}), + # Durable finalization can run on the one unscoped stale-monitor thread; + # retain the dispatching profile so that thread updates the same state.db. + "_profile_home": str(get_hermes_home()), # Stale-monitor bookkeeping (see _stale_monitor_loop). "_progress_token": None, "_progress_ts": dispatched_at, "_interrupted_at": None} with _records_lock: @@ -724,11 +727,19 @@ def _push_completion_event(record: Dict[str, Any], result: Dict[str, Any], statu **({} if is_batch else {"exit_reason": result.get("exit_reason")}), **{k: record[k] for k in _ROUTING_KEYS if record.get(k)}, **{k: result[k] for k in _STALL_META_KEYS if k in result}} + token = None try: + profile_home = record.get("_profile_home") + if profile_home: + from hermes_constants import reset_hermes_home_override, set_hermes_home_override + token = set_hermes_home_override(profile_home) _persist_completion(evt, result) except Exception as exc: # noqa: BLE001 — a lost durable row is recoverable; a lost result + leaked slot is not logger.error(f"Async delegation{label} %s: durable completion write failed; delivering in-memory " "only (a restart may report this unit as unknown): %s", record.get("delegation_id"), exc) + finally: + if token is not None: + reset_hermes_home_override(token) try: process_registry.completion_queue.put(evt) except Exception as exc: # pragma: no cover