fix(agent-server): make drain_reader_threads async to prevent event-loop deadlock (#657)

.claude/agents/test-runner.md

@@ -172,6 +172,7 @@ The test suite covers:
 - **Local Deployment** (test_deploy_local.py) - Deploy local agents via MCP (Req 11.2)
 - **Public Links** (test_public_links.py) - Public agent sharing; email verification is driven by agent-level access policy (#311 follow-up, 2026-04-13) (Req 11.3)
 - **Public User Memory** (test_public_user_memory.py) - Per-user persistent memory for email-verified public sessions (MEM-001) [SMOKE]
+- **Site Proxy** (test_site_proxy.py) - Agent website proxy via /site/{token}/{path}; link_type field, URL format, token validation, 502 on no web server, redirect (SITE-001, #633)
 
 ### Operations & Observability
 - **Fleet Operations** (test_ops.py) - Fleet status/health, restart/stop, schedule list/pause/resume, emergency stop, alerts, costs [SLOW]
@@ -233,7 +234,7 @@ The test suite covers:
 
 ## Test Suite Statistics
 
-**Total Tests**: ~2,262 tests across 124 test files
+**Total Tests**: ~2,271 tests across 125 test files
 **Smoke Tests**: ~578 tests (fast, no agent creation)
 **Unit Tests**: ~170 tests (no backend needed, rate limit detection, watchdog logic, context formula, OTel trace logging, file upload, voice transcription, inter-agent timeout, scheduler sync loop, team-share gate, WhatsApp adapter (67), subprocess pgroup (11), empty result classification (13))
 **Core Tests (not slow)**: ~2,112 tests
@@ -265,6 +266,14 @@ Use these thresholds to assess test health (based on **executed** tests, not inc
 - **Warning**: 75-90% pass rate, <5 failures
 - **Critical**: <75% pass rate or >5 failures
 
+## Recent Test Additions (2026-05-05)
+
+| Test File | Description | Tests Added |
+|-----------|-------------|-------------|
+| `unit/test_subprocess_pgroup.py` | #657 — `drain_reader_threads` async refactor: updated all 5 `drain_reader_threads(...)` call sites to `asyncio.run(drain_reader_threads(...))` following the function being converted to `async def`. All 12 existing tests pass. No new tests required — the existing `TestDrainOrphanKillerTimeout.test_drain_bounded_when_orphan_killer_blocks` was already the regression test for the deadlock scenario; after the fix it completes in ~14s instead of 100s. | 0 new, 12 updated call sites |
+
+---
+
 ## Recent Test Additions (2026-05-04)
 
 | Test File | Description | Tests Added |
@@ -275,6 +284,12 @@ Use these thresholds to assess test health (based on **executed** tests, not inc
 
 Root cause of all `POST/GET/DELETE /api/agents/{name}/schedules/{id}/webhook` and `POST /api/webhooks/{token}` returning 500 on a live stack since #291 (Nov 2025). WEBHOOK-001 added the four methods to `ScheduleOperations` but never added the matching pass-throughs on the `DatabaseManager` facade — and there's no `__getattr__` proxy. The integration test from PR #643 would have caught it but doesn't run in CI. New unit test runs in any Python (no live stack), would have caught the regression at PR-time.
 
+## Recent Test Additions (2026-05-03)
+
+| Test File | Description | Tests Added |
+|-----------|-------------|-------------|
+| `test_site_proxy.py` | SITE-001 (#633) — agent website proxy: `test_create_chat_link_default_type`, `test_create_site_link` (URL uses `/site/`), `test_create_invalid_link_type_rejected`, `test_list_links_includes_type`, `test_invalid_token_returns_401`, `test_chat_token_rejected_at_site_endpoint`, `test_valid_site_token_502_when_agent_not_running`, `test_disabled_link_returns_401`, `test_root_path_without_trailing_slash_redirects` | 9 new tests |
+
 ---
 
 ## Recent Test Additions (2026-04-27)

README.md

@@ -1,4 +1,6 @@
 <div align="center">
+  <a href="https://www.ability.ai/workshopsgt"><strong>🛠️ Free Weekly Workshops — Learn to build AI agents on Trinity →</strong></a>
+  <br/><br/>
   <img src="docs/assets/trinity-logo-white.svg#gh-dark-mode-only" alt="Trinity" width="120"/>
   <img src="docs/assets/trinity-logo.svg#gh-light-mode-only" alt="Trinity" width="120"/>
   <h1>Trinity</h1>

docker/base-image/agent_server/services/claude_code.py

@@ -950,16 +950,16 @@ def read_subprocess_output():
                     f"— killing process group"
                 )
                 _terminate_process_group(process, graceful_timeout=5, pgid=process_pgid)
-                _drain_reader_threads(
+                asyncio.run(_drain_reader_threads(
                     process, stdout_thread, stderr_thread,
                     grace=3, pgid=process_pgid,
-                )
+                ))
                 raise
 
-            _drain_reader_threads(
+            asyncio.run(_drain_reader_threads(
                 process, stdout_thread, stderr_thread,
                 grace=5, pgid=process_pgid,
-            )
+            ))
 
             stderr = ''.join(stderr_lines)
             stderr = sanitize_text(stderr) if stderr else stderr
@@ -1743,19 +1743,19 @@ def read_subprocess_output_with_timeout():
                     f"— killing process group"
                 )
                 _terminate_process_group(process, graceful_timeout=5, pgid=process_pgid)
-                _drain_reader_threads(
+                asyncio.run(_drain_reader_threads(
                     process, stdout_thread, stderr_thread,
                     grace=3, pgid=process_pgid,
-                )
+                ))
                 raise
 
             # Subprocess exited. Drain readers — if a hook grandchild still
             # holds a pipe, the helper will close the pipe FDs so the
             # reader threads can exit.
-            _drain_reader_threads(
+            asyncio.run(_drain_reader_threads(
                 process, stdout_thread, stderr_thread,
                 grace=5, pgid=process_pgid,
-            )
+            ))
 
             # Re-raise permission-mode failure captured by stdout thread
             if stdout_exc:

docker/base-image/agent_server/utils/subprocess_pgroup.py

@@ -24,6 +24,7 @@
 """
 from __future__ import annotations
 
+import asyncio
 import logging
 import os
 import signal
@@ -205,7 +206,7 @@ def _kill_orphan_pipe_writers(pipe_read_fd: int, our_pgid: Optional[int]) -> int
     return killed
 
 
-def drain_reader_threads(
+async def drain_reader_threads(
     process: subprocess.Popen,
     *threads: Optional[threading.Thread],
     grace: int = 5,
@@ -230,10 +231,24 @@ def drain_reader_threads(
     Callers that have already reaped the parent via ``process.wait()``
     must pass ``pgid`` — after reaping, the pid is gone and we'd
     otherwise lose the ability to signal grandchildren.
+
+    This function is ``async`` so every ``t.join()`` runs off the event-loop
+    thread via ``asyncio.to_thread``.  ``asyncio.wait_for`` enforces the
+    deadline at the event-loop level, preventing the asyncio event loop from
+    blocking even when orphan subprocesses consume all available CPUs and
+    starve ordinary OS-level thread timeouts.  Sync callers (e.g. executor
+    thread functions) must wrap this with ``asyncio.run(drain_reader_threads(…))``.
+    Issue #657.
     """
     alive_threads = [t for t in threads if t is not None]
+
+    # Initial grace-period joins — each runs in a worker thread so the event
+    # loop enforces the deadline even under CPU pressure from orphan processes.
     for t in alive_threads:
-        t.join(timeout=grace)
+        try:
+            await asyncio.wait_for(asyncio.to_thread(t.join, grace), timeout=grace + 1)
+        except asyncio.TimeoutError:
+            pass
 
     stuck = [t for t in alive_threads if t.is_alive()]
     if not stuck:
@@ -256,29 +271,41 @@ def drain_reader_threads(
     # our reader.  Killing it releases all its FDs (stdout AND stderr write
     # ends), unblocking both reader threads simultaneously.
     #
-    # Issue #649: run in a daemon thread with a hard 10-second timeout.
-    # Iterating /proc can block indefinitely when a process is in D state
-    # (uninterruptible sleep), causing the drain to stall for tens of minutes
-    # instead of the expected ~30 seconds.
+    # Issue #649 / #657: the /proc scan runs in a daemon thread.  We wait on
+    # a threading.Event (via asyncio.to_thread) rather than joining the thread
+    # directly — event.wait(10) always returns within 10 seconds, so
+    # asyncio.run()'s shutdown_default_executor() never blocks on a slow scan.
     if process.stdout is not None and not process.stdout.closed:
         try:
             stdout_fd = process.stdout.fileno()
         except Exception:
             stdout_fd = None
         if stdout_fd is not None:
             _orphan_result: list[int] = [0]
+            _orphan_done = threading.Event()
 
             def _run_orphan_killer() -> None:
                 try:
                     _orphan_result[0] = _kill_orphan_pipe_writers(stdout_fd, pgid)
                 except Exception:
                     pass  # best-effort; never fail the drain path
+                finally:
+                    _orphan_done.set()
 
-            _orphan_thread = threading.Thread(target=_run_orphan_killer, daemon=True)
-            _orphan_thread.start()
-            _orphan_thread.join(timeout=10)
+            threading.Thread(target=_run_orphan_killer, daemon=True).start()
+
+            # asyncio.to_thread(event.wait, 10) runs event.wait(10) in the
+            # executor; it always returns within 10s so asyncio.run()
+            # shutdown_default_executor() exits promptly even when the
+            # daemon thread is still scanning /proc.
+            try:
+                await asyncio.wait_for(
+                    asyncio.to_thread(_orphan_done.wait, 10), timeout=11
+                )
+            except asyncio.TimeoutError:
+                pass
 
-            if _orphan_thread.is_alive():
+            if not _orphan_done.is_set():
                 logger.warning(
                     "[Subprocess] _kill_orphan_pipe_writers still running after "
                     "10s (pid=%s) — /proc scan may be blocked on a D-state process",
@@ -298,8 +325,16 @@ def _run_orphan_killer() -> None:
     # post_kill_grace, clamp to 1 second so we still attempt a natural drain.
     elapsed = time.monotonic() - drain_start
     join_timeout = max(1.0, post_kill_grace - elapsed)
+
+    # Natural-drain joins — run off the event loop so asyncio.wait_for enforces
+    # the deadline even when the reader thread is blocked by a CPU-heavy orphan.
     for t in stuck:
-        t.join(timeout=join_timeout)
+        try:
+            await asyncio.wait_for(
+                asyncio.to_thread(t.join, join_timeout), timeout=join_timeout + 1
+            )
+        except asyncio.TimeoutError:
+            pass
 
     still_stuck = [t for t in stuck if t.is_alive()]
     if not still_stuck:
@@ -320,8 +355,12 @@ def _run_orphan_killer() -> None:
         elapsed, process.pid, len(still_stuck),
     )
     safe_close_pipes(process)
+
     for t in still_stuck:
-        t.join(timeout=2)
+        try:
+            await asyncio.wait_for(asyncio.to_thread(t.join, 2), timeout=3)
+        except asyncio.TimeoutError:
+            pass
 
     leaked = [t for t in still_stuck if t.is_alive()]
     if leaked:

docs/memory/feature-flows/parallel-headless-execution.md

@@ -3,13 +3,16 @@
 > **Requirement**: 12.1 - Parallel Headless Execution
 > **Status**: Implemented
 > **Created**: 2025-12-22
-> **Updated**: 2026-04-27 (#531: `drain_reader_threads` reordered to preserve kernel pipe buffer before close)
+> **Updated**: 2026-05-05 (#657: `drain_reader_threads` made async; all `t.join()` calls offloaded via `asyncio.to_thread` + `asyncio.wait_for`)
 > **Verified**: 2026-02-05
 
 ## Revision History
 
 | Date | Changes |
 |------|---------|
+| 2026-05-05 | **Issue #657 - `drain_reader_threads` `t.join(30)` deadlocks asyncio event loop permanently when reader thread is stuck**: When `drain_reader_threads` is called and a CPU-heavy orphan MCP subprocess (that called `setsid()`) survives the process-group kill, the orphan consumes 140–180% CPU, starving the asyncio event loop and potentially preventing OS-level thread timeouts from firing. All three `t.join(timeout=N)` calls in `drain_reader_threads` — initial grace join, natural-drain join, and post-force-close join — as well as the orphan-killer `_orphan_thread.join(timeout=10)` could block indefinitely. Fix (`subprocess_pgroup.py`): convert `drain_reader_threads` to `async def`. Replace all `t.join(timeout=N)` with `await asyncio.wait_for(asyncio.to_thread(t.join, N), timeout=N+1)`. Deadlines are now enforced by the asyncio event loop, not by OS thread scheduling. The orphan killer changes from `_orphan_thread.join(timeout=10)` to: start daemon thread, signal a `threading.Event` in a `finally` block, then `await asyncio.wait_for(asyncio.to_thread(event.wait, 10), timeout=11)` — the `event.wait(10)` always returns within 10s so `asyncio.run()`'s `shutdown_default_executor()` never blocks on the slow `/proc` scan. Sync callers in `claude_code.py` (both the chat and headless-task paths, 4 call sites) wrap with `asyncio.run(drain_reader_threads(...))`, which creates a temporary event loop in the executor thread. Tests updated to call `asyncio.run(drain_reader_threads(...))` — all 12 `TestDrainReaderThreads` / `TestDrainOrphanKillerTimeout` tests pass; `test_drain_bounded_when_orphan_killer_blocks` now completes in ~14s instead of 100s. Companion to #649 (orphan killer 10s cap). |
+| 2026-05-04 | **Issue #649 - `_kill_orphan_pipe_writers` blocks drain indefinitely on D-state `/proc` entries**: When `drain_reader_threads` falls into the post-kill path (reader still alive after `post_kill_grace`), it calls `_kill_orphan_pipe_writers` inline on the calling thread to kill npx/npm MCP servers that survived `terminate_process_group`. If any process in `/proc` is in D state (uninterruptible sleep — e.g. an NFS hang), `os.listdir(f"/proc/{pid}/fd")` blocks without a timeout, stalling the drain for tens of minutes instead of the expected ~30s. Fix (`subprocess_pgroup.py`): wrap the `_kill_orphan_pipe_writers` call in a daemon thread with `join(timeout=10)`. If the thread is still alive after 10s, log a warning and proceed — the orphan scan is best-effort. Also add wall-clock accounting: record `drain_start = time.monotonic()` when the WARNING fires, then compute `join_timeout = max(1.0, post_kill_grace - elapsed)` so the remaining budget for the reader join is measured from the same wall-clock origin, not from after the orphan scan. The combination means the total post-kill phase is bounded by `max(10, post_kill_grace)` seconds regardless of `/proc` behavior. Regression test: `TestDrainOrphanKillerTimeout.test_drain_bounded_when_orphan_killer_blocks` in `tests/unit/test_subprocess_pgroup.py` — monkeypatches `_kill_orphan_pipe_writers` to sleep 100s and verifies drain completes in < 20s (10s orphan cap + 5s post_kill_grace). |
+| 2026-04-27 (backfill) | **Issue #618 - npx MCP server orphans outside claude pgid hold stdout pipe open**: `terminate_process_group(claude_pgid)` kills only processes inside the claude session's pgid. npm → node chains started by `npx`-based MCP servers call `setsid()` on the child, placing it in a new pgid that `os.killpg(claude_pgid, ...)` never touches. The orphan keeps the pipe write-FD open so the kernel never delivers EOF to the reader thread. Fix: after `terminate_process_group` in `drain_reader_threads`, scan `/proc/{pid}/fd` for any write-end holder of the same stdout pipe inode that is *outside* our pgid and send it `SIGKILL`. Identity confirmed via `fstat.st_ino` match + `fdinfo flags` write-bit check. Returns kill count for logging. Added `_kill_orphan_pipe_writers(pipe_read_fd, our_pgid) -> int` to `subprocess_pgroup.py`. (This fix ships as part of the #649 PR since both live in `subprocess_pgroup.py`; #618 is the motivation, #649 is the safety cap.) |
 | 2026-04-27 | **Issue #531 - drain_reader_threads closes stdout pipe before reader can drain backlog**: Root cause of the HTTP 502 "Execution completed without a result message" on long agentic tasks (>5 min, dozens of tool calls). In `drain_reader_threads` (`subprocess_pgroup.py`), the old sequence was: kill grandchildren → `safe_close_pipes()` → join reader (2s). Closing the read end of the pipe before the reader finished discarded the kernel pipe buffer — including the final `{"type":"result"}` JSON line written by claude just before exit. This caused `metadata.cost_usd` and `metadata.duration_ms` to stay `None`, which #521's `_classify_empty_result` correctly identified as a 502. Fix: reorder to kill grandchildren → join reader with `post_kill_grace=30s` (natural drain: grandchildren dead → kernel delivers EOF → reader sees buffered tail → exits cleanly) → `safe_close_pipes()` only as a last resort for genuine wedges. Also: `_classify_empty_result(metadata, raw_message_count, raw_messages)` now accepts `raw_messages` list and derives `num_turns` from it when `metadata.num_turns is None` (which is the case when the result line is lost), so the 502 detail shows an honest turn count. `metadata.tool_count` is accumulated per-message during parsing (not from the result line) and remains reliable. New parameter `post_kill_grace=30` added to `drain_reader_threads`; all call sites use the default. Regression test: `test_buffered_data_preserved_after_grandchild_kill` in `tests/unit/test_subprocess_pgroup.py`. Related: #520 (symptom), #521 (502 classification), #523 (FD_CLOEXEC complement, future). |
 | 2026-04-26 | **Issue #520 - Empty-result misreported as success**: When the claude subprocess exits cleanly (`return_code == 0`) but the final `{"type":"result"}` JSON line is dropped before the reader thread captures it (typical cause: a child subprocess inherited stdout, kept the pipe open past claude exit, the reader thread leaked, the pgroup unwind closed the pipe and the result line went with it), `metadata.cost_usd` and `metadata.duration_ms` stay `None`. The success path used to return HTTP 200 anyway — agent-server logged "completed successfully" while backend silently reaped the execution as an orphan minutes later, masking the real failure with a misleading "completed on agent but recovered by watchdog" message. Sibling of #516 (signal-kill path). Added `_classify_empty_result(metadata, raw_message_count)` (`claude_code.py:935-980`) consulted *after* the `return_code != 0` block (#516 + auth heuristics) and *before* response building; when both `cost_usd` and `duration_ms` are `None`, raises HTTP 502 with diagnostic context (tool count, turn count, raw_messages, cause hint). Backend's auth classifier only triggers on 503, so 502 flows through as plain FAILED with the helpful detail preserved — no backend changes. The two-field check is conservative: single-field nullability could be a Claude format quirk; both-None is a strong signal that the terminal `result` message never arrived. 9 unit tests in `tests/unit/test_empty_result_classification.py`. |
 | 2026-04-26 | **Issue #516 - Signal kills misclassified as auth failure**: External signal terminations of the claude subprocess (timeout SIGKILL, OOM-kill, parent SIGTERM, operator cancel) used to fall into the auth-fallback heuristics at `claude_code.py:1262-1278` and surface as a misleading "Subscription token may be expired" 503 — same shape as #361 but a different exit path. Added `_classify_signal_exit(return_code, metadata)` helper (`claude_code.py:882-931`) consulted *before* the auth heuristics; it matches Python-native signal exits (`return_code < 0`) and shell-encoded forms (`130, 137, 143` for SIGINT/SIGKILL/SIGTERM) and raises HTTP 504 with a clear "killed by SIGKILL/SIGTERM/SIGINT — likely timeout, OOM, or operator cancel" message. Tightened the zero-token heuristic with `return_code > 0` so signal exits cannot reach it. Backend's `task_execution_service.py:542` only flags AUTH on 503, so 504 falls through to the generic FAILED path — no backend changes needed. The bug became routinely reproducible after #61 (PR #326) added backend-driven `terminate_execution_on_agent()` which makes signal-killed claude subprocesses the common case for any timeout. 21 unit tests in `tests/unit/test_signal_exit_classification.py`. |