fix(runtime): create background job tasks on the loop thread

hermes_progress_tail/runtime/clarify.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import json
 import logging
 import re
@@ -11,7 +10,7 @@
 
 from ..models.decision import CheckpointStatus
 from ..utils.redaction import redact_text
-from ..utils.scheduling import create_task_or_close, submit_or_close
+from ..utils.scheduling import create_task_or_close, on_loop_thread, submit_or_close
 
 logger = logging.getLogger(__name__)
 
@@ -38,13 +37,6 @@ def _usable_loop(ctx: Any) -> Any | None:
     return loop
 
 
-def _on_loop(loop: Any) -> bool:
-    try:
-        return asyncio.get_running_loop() is loop
-    except RuntimeError:
-        return False
-
-
 def _diagnostic(label: str, exc: BaseException | str) -> None:
     text = _CONTROL.sub(" ", redact_text(str(exc)))
     logger.debug("hermes-progress-tail clarify %s: %s", label, " ".join(text.split())[:160])
@@ -69,7 +61,7 @@ def run_clarify_freeze_barrier(
         return False
     guard = FreezeAttemptGuard(threading.Event())
     try:
-        if _on_loop(loop):
+        if on_loop_thread(loop):
             create_task_or_close(loop, operation(guard)).add_done_callback(
                 _consume("same-loop freeze failed")
             )
@@ -90,7 +82,7 @@ def run_clarify_resolution_bridge(ctx: Any, operation: Callable[[], Awaitable[bo
     if loop is None:
         return
     try:
-        if _on_loop(loop):
+        if on_loop_thread(loop):
             create_task_or_close(loop, operation()).add_done_callback(
                 _consume("same-loop resolution failed")
             )

hermes_progress_tail/runtime/tool_events.py

@@ -18,7 +18,7 @@
     ToolEvent,
 )
 from ..rendering.formatter import extract_todo_items, format_tool_line
-from ..utils.scheduling import create_task_or_close, submit_or_close
+from ..utils.scheduling import create_task_or_close, schedule_task_on_loop, submit_or_close
 from .clarify import (
     classify_clarify_result,
     run_clarify_freeze_barrier,
@@ -199,7 +199,7 @@ async def _cleanup() -> None:
             force=True,
         )
 
-    create_task_or_close(ctx.loop, _cleanup())
+    schedule_task_on_loop(ctx.loop, lambda: create_task_or_close(ctx.loop, _cleanup()))
 
 
 def _schedule_background_job_poll(ctx: SessionContext, process_id: str) -> None:
@@ -210,11 +210,11 @@ def _schedule_background_job_poll(ctx: SessionContext, process_id: str) -> None:
         or not runtime_plugin.get_renderer().settings.background_jobs.enabled
     ):
         return
-    job = ctx.background.jobs.get(process_id)
-    if job is not None and job.poll_task is not None and not job.poll_task.done():
-        return
 
     async def _poll() -> None:
+        current = ctx.background.jobs.get(process_id)
+        if current is not None and current.poll_task is None:
+            current.poll_task = asyncio.current_task()
         try:
             while True:
                 await asyncio.sleep(
@@ -267,10 +267,17 @@ async def _poll() -> None:
         except Exception:
             logger.debug("hermes-progress-tail background job poll failed", exc_info=True)
 
-    task = create_task_or_close(ctx.loop, _poll())
-    job = ctx.background.jobs.get(process_id)
-    if job is not None:
-        job.poll_task = task
+    def _start() -> asyncio.Task[Any]:
+        job = ctx.background.jobs.get(process_id)
+        if job is not None and job.poll_task is not None and not job.poll_task.done():
+            return job.poll_task
+        task = create_task_or_close(ctx.loop, _poll())
+        job = ctx.background.jobs.get(process_id)
+        if job is not None:
+            job.poll_task = task
+        return task
+
+    schedule_task_on_loop(ctx.loop, _start)
 
 
 def _is_context_owner_thread(ctx: SessionContext) -> bool:

hermes_progress_tail/utils/scheduling.py

@@ -1,11 +1,18 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Coroutine
+from collections.abc import Callable, Coroutine
 from concurrent.futures import Future
 from typing import Any
 
 
+def on_loop_thread(loop: asyncio.AbstractEventLoop) -> bool:
+    try:
+        return asyncio.get_running_loop() is loop
+    except RuntimeError:
+        return False
+
+
 def submit_or_close(coro: Coroutine[Any, Any, Any], loop: asyncio.AbstractEventLoop) -> Future[Any]:
     try:
         return asyncio.run_coroutine_threadsafe(coro, loop)
@@ -22,3 +29,12 @@ def create_task_or_close(
     except BaseException:
         coro.close()
         raise
+
+
+def schedule_task_on_loop(
+    loop: asyncio.AbstractEventLoop, start: Callable[[], asyncio.Task[Any]]
+) -> asyncio.Task[Any] | None:
+    if on_loop_thread(loop):
+        return start()
+    loop.call_soon_threadsafe(start)
+    return None

tests/test_runtime_background_process_polling.py

@@ -16,6 +16,9 @@ def create_task(self, coroutine):
         self.coroutines.append(coroutine)
         return SimpleNamespace(done=lambda: False)
 
+    def call_soon_threadsafe(self, callback):
+        callback()
+
 
 class Registry:
     def __init__(self, values):

tests/test_thread_safe_scheduling.py

@@ -0,0 +1,188 @@
+import asyncio
+import sys
+import threading
+import time
+from types import ModuleType, SimpleNamespace
+
+from hermes_progress_tail.models.state import BackgroundJob, SessionContext
+from hermes_progress_tail.renderer import ProgressRenderer
+from hermes_progress_tail.runtime import plugin, tool_events
+from hermes_progress_tail.settings.loading import load_settings
+from tests.support.rendering import EditableAdapter
+
+
+class LoopThread:
+    def __init__(self):
+        self.loop = asyncio.new_event_loop()
+        self.loop.set_debug(True)
+        self.thread = threading.Thread(target=self.loop.run_forever, daemon=True)
+
+    def __enter__(self):
+        self.thread.start()
+        return self
+
+    def __exit__(self, *_):
+        async def shutdown():
+            pending = [task for task in asyncio.all_tasks() if task is not asyncio.current_task()]
+            for task in pending:
+                task.cancel()
+            await asyncio.gather(*pending, return_exceptions=True)
+
+        asyncio.run_coroutine_threadsafe(shutdown(), self.loop).result(5)
+        self.loop.call_soon_threadsafe(self.loop.stop)
+        self.thread.join(5)
+        self.loop.close()
+
+
+class Registry:
+    def __init__(self, values):
+        self.values = iter(values)
+        self.pending_watchers = []
+
+    def get(self, process_id):
+        return next(self.values)
+
+
+class Session:
+    def __init__(self):
+        self.output_buffer = "running"
+        self.exited = False
+        self.exit_code = None
+        self.command = "pytest -q"
+        self.cwd = "/repo"
+        self.pid = 9
+        self.notify_on_complete = True
+        self.watcher_interval = 10
+        self.watch_patterns = ["done"]
+
+    def complete(self):
+        self.output_buffer = "2 passed"
+        self.exited = True
+        self.exit_code = 0
+
+
+class CompletingRegistry(Registry):
+    def __init__(self, session, *, after=2):
+        super().__init__([])
+        self.session = session
+        self.after = after
+        self.reads = 0
+
+    def get(self, process_id):
+        self.reads += 1
+        if self.reads >= self.after:
+            self.session.complete()
+        return self.session
+
+
+def _install_registry(monkeypatch, registry):
+    package = ModuleType("tools")
+    module = ModuleType("tools.process_registry")
+    module.process_registry = registry
+    monkeypatch.setitem(sys.modules, "tools", package)
+    monkeypatch.setitem(sys.modules, "tools.process_registry", module)
+
+
+def _ctx(loop):
+    return SessionContext("sid", "key", "discord", "chat", None, None, loop)
+
+
+def _renderer(**job_settings):
+    defaults = {"enabled": True, "update_interval_seconds": 0, "completed_ttl_seconds": 0}
+    defaults.update(job_settings)
+    return SimpleNamespace(settings=SimpleNamespace(background_jobs=SimpleNamespace(**defaults)))
+
+
+def test_poll_from_a_foreign_thread_runs_on_an_idle_loop(monkeypatch):
+    events = []
+    delivered = threading.Event()
+
+    def record(context, event, force=False):
+        events.append(event)
+        delivered.set()
+
+    monkeypatch.setattr(tool_events, "_schedule_render", record)
+    monkeypatch.setattr(plugin, "_get_renderer", lambda: _renderer())
+    _install_registry(monkeypatch, Registry([None]))
+
+    with LoopThread() as harness:
+        ctx = _ctx(harness.loop)
+        ctx.background.jobs["proc"] = BackgroundJob("proc")
+        tool_events._schedule_background_job_poll(ctx, "proc")
+        assert delivered.wait(5)
+
+    assert [(event.process_id, event.event_type, event.exited) for event in events] == [
+        ("proc", "lost", True)
+    ]
+
+
+def test_cleanup_from_a_foreign_thread_runs_on_an_idle_loop(monkeypatch):
+    events = []
+    delivered = threading.Event()
+
+    def record(context, event, force=False):
+        events.append((event, force))
+        delivered.set()
+
+    monkeypatch.setattr(tool_events, "_schedule_render", record)
+    monkeypatch.setattr(plugin, "_get_renderer", lambda: _renderer())
+
+    with LoopThread() as harness:
+        ctx = _ctx(harness.loop)
+        tool_events._schedule_background_job_cleanup(ctx, "proc")
+        assert delivered.wait(5)
+
+    assert [(event.process_id, event.event_type, force) for event, force in events] == [
+        ("proc", "cleanup", True)
+    ]
+
+
+def test_post_tool_hook_from_a_worker_thread_polls_without_another_loop_event(monkeypatch):
+    adapter = EditableAdapter()
+    renderer = ProgressRenderer(
+        load_settings(
+            {
+                "progress_tail": {
+                    "tools": {"timestamp": False},
+                    "background_jobs": {"update_interval_seconds": 1},
+                }
+            }
+        )
+    )
+    _install_registry(monkeypatch, CompletingRegistry(Session()))
+    monkeypatch.setattr(plugin, "_get_renderer", lambda: renderer)
+
+    with LoopThread() as harness:
+        ctx = SessionContext(
+            "s1",
+            "k1",
+            "discord",
+            "chat",
+            None,
+            adapter,
+            harness.loop,
+            "live_tail",
+            timestamp=False,
+        )
+        renderer.register_context(ctx)
+        plugin._on_post_tool_call(
+            "terminal",
+            args={"background": True, "command": "pytest -q"},
+            result='{"session_id":"proc_live","exited":false}',
+            task_id="k1",
+            session_id="s1",
+        )
+        deadline = time.monotonic() + 10
+        claimed = False
+        while time.monotonic() < deadline:
+            job = ctx.background_jobs.get("proc_live")
+            if job is not None and job.status == "running" and job.poll_task is not None:
+                claimed = True
+            if job is not None and job.status == "completed":
+                break
+            time.sleep(0.01)
+        job = ctx.background_jobs["proc_live"]
+
+    assert claimed
+    assert (job.status, job.exit_code) == ("completed", 0)
+    assert "2 passed" in " ".join(job.output_tail)