common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: background process monitoring — watch_patterns for real-time output alerts

Browse Source 
* feat: add watch_patterns to background processes for output monitoring

Adds a new 'watch_patterns' parameter to terminal(background=true) that
lets the agent specify strings to watch for in process output. When a
matching line appears, a notification is queued and injected as a
synthetic message — triggering a new agent turn, similar to
notify_on_complete but mid-process.

Implementation:
- ProcessSession gets watch_patterns field + rate-limit state
- _check_watch_patterns() in ProcessRegistry scans new output chunks
  from all three reader threads (local, PTY, env-poller)
- Rate limited: max 8 notifications per 10s window
- Sustained overload (45s) permanently disables watching for that process
- watch_queue alongside completion_queue, same consumption pattern
- CLI drains watch_queue in both idle loop and post-turn drain
- Gateway drains after agent runs via _inject_watch_notification()
- Checkpoint persistence + crash recovery includes watch_patterns
- Blocked in execute_code sandbox (like other bg params)
- 20 new tests covering matching, rate limiting, overload kill,
  checkpoint persistence, schema, and handler passthrough

Usage:
  terminal(
      command='npm run dev',
      background=true,
      watch_patterns=['ERROR', 'WARN', 'listening on port']
  )

* refactor: merge watch_queue into completion_queue

Unified queue with 'type' field distinguishing 'completion',
'watch_match', and 'watch_disabled' events. Extracted
_format_process_notification() in CLI and gateway to handle
all event types in a single drain loop. Removes duplication
across both CLI drain sites and the gateway.
This commit is contained in:
Teknium
2026-04-11 03:13:23 -07:00
committed by GitHub
parent a2f9f04c06
commit f459214010
6 changed files with 556 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tools/code_execution_tool.py
View File

@@ -301,7 +301,7 @@ def _call(tool_name, args):
# ---------------------------------------------------------------------------
# Terminal parameters that must not be used from ephemeral sandbox scripts
_TERMINAL_BLOCKED_PARAMS = {"background", "check_interval", "pty", "notify_on_complete"}
_TERMINAL_BLOCKED_PARAMS = {"background", "check_interval", "pty", "notify_on_complete", "watch_patterns"}
def _rpc_server_loop(

109
tools/process_registry.py
View File

@@ -58,6 +58,11 @@ MAX_OUTPUT_CHARS = 200_000 # 200KB rolling output buffer
FINISHED_TTL_SECONDS = 1800 # Keep finished processes for 30 minutes
MAX_PROCESSES = 64 # Max concurrent tracked processes (LRU pruning)
# Watch pattern rate limiting
WATCH_MAX_PER_WINDOW = 8 # Max notifications delivered per window
WATCH_WINDOW_SECONDS = 10 # Rolling window length
WATCH_OVERLOAD_KILL_SECONDS = 45 # Sustained overload duration before disabling watch
@dataclass
class ProcessSession:
@@ -83,6 +88,14 @@ class ProcessSession:
watcher_thread_id: str = ""
watcher_interval: int = 0 # 0 = no watcher configured
notify_on_complete: bool = False # Queue agent notification on exit
# Watch patterns — trigger agent notification when output matches any pattern
watch_patterns: List[str] = field(default_factory=list)
_watch_hits: int = field(default=0, repr=False) # total matches delivered
_watch_suppressed: int = field(default=0, repr=False) # matches dropped by rate limit
_watch_overload_since: float = field(default=0.0, repr=False) # when sustained overload began
_watch_disabled: bool = field(default=False, repr=False) # permanently killed by overload
_watch_window_hits: int = field(default=0, repr=False) # hits in current rate window
_watch_window_start: float = field(default=0.0, repr=False)
_lock: threading.Lock = field(default_factory=threading.Lock)
_reader_thread: Optional[threading.Thread] = field(default=None, repr=False)
_pty: Any = field(default=None, repr=False) # ptyprocess handle (when use_pty=True)
@@ -114,9 +127,10 @@ class ProcessRegistry:
# Side-channel for check_interval watchers (gateway reads after agent run)
self.pending_watchers: List[Dict[str, Any]] = []
# Completion notifications — processes with notify_on_complete push here
# on exit. CLI process_loop and gateway drain this after each agent turn
# to auto-trigger a new agent turn with the process results.
# Notification queue — unified queue for all background process events.
# Completion notifications (notify_on_complete) and watch pattern matches
# both land here, distinguished by "type" field. CLI process_loop and
# gateway drain this after each agent turn to auto-trigger new turns.
import queue as _queue_mod
self.completion_queue: _queue_mod.Queue = _queue_mod.Queue()
@@ -128,6 +142,84 @@ class ProcessRegistry:
lines.pop(0)
return "\n".join(lines)
def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None:
"""Scan new output for watch patterns and queue notifications.
Called from reader threads with new_text being the freshly-read chunk.
Rate-limited: max WATCH_MAX_PER_WINDOW notifications per WATCH_WINDOW_SECONDS.
If sustained overload exceeds WATCH_OVERLOAD_KILL_SECONDS, watching is
disabled permanently for this process.
"""
if not session.watch_patterns or session._watch_disabled:
return
# Scan new text line-by-line for pattern matches
matched_lines = []
matched_pattern = None
for line in new_text.splitlines():
for pat in session.watch_patterns:
if pat in line:
matched_lines.append(line.rstrip())
if matched_pattern is None:
matched_pattern = pat
break # one match per line is enough
if not matched_lines:
return
now = time.time()
with session._lock:
# Reset window if it's expired
if now - session._watch_window_start >= WATCH_WINDOW_SECONDS:
session._watch_window_hits = 0
session._watch_window_start = now
# Check rate limit
if session._watch_window_hits >= WATCH_MAX_PER_WINDOW:
session._watch_suppressed += len(matched_lines)
# Track sustained overload for kill switch
if session._watch_overload_since == 0.0:
session._watch_overload_since = now
elif now - session._watch_overload_since > WATCH_OVERLOAD_KILL_SECONDS:
session._watch_disabled = True
self.completion_queue.put({
"session_id": session.id,
"command": session.command,
"type": "watch_disabled",
"suppressed": session._watch_suppressed,
"message": (
f"Watch patterns disabled for process {session.id} — "
f"too many matches ({session._watch_suppressed} suppressed). "
f"Use process(action='poll') to check output manually."
),
})
return
# Under the rate limit — deliver notification
session._watch_window_hits += 1
session._watch_hits += 1
# Clear overload tracker since we got a delivery through
session._watch_overload_since = 0.0
# Include suppressed count if any events were dropped
suppressed = session._watch_suppressed
session._watch_suppressed = 0
# Trim matched output to a reasonable size
output = "\n".join(matched_lines[:20])
if len(output) > 2000:
output = output[:2000] + "\n...(truncated)"
self.completion_queue.put({
"session_id": session.id,
"command": session.command,
"type": "watch_match",
"pattern": matched_pattern,
"output": output,
"suppressed": suppressed,
})
@staticmethod
def _is_host_pid_alive(pid: Optional[int]) -> bool:
"""Best-effort liveness check for host-visible PIDs."""
@@ -394,6 +486,7 @@ class ProcessRegistry:
session.output_buffer += chunk
if len(session.output_buffer) > session.max_output_chars:
session.output_buffer = session.output_buffer[-session.max_output_chars:]
self._check_watch_patterns(session, chunk)
except Exception as e:
logger.debug("Process stdout reader ended: %s", e)
finally:
@@ -413,6 +506,7 @@ class ProcessRegistry:
quoted_log_path = shlex.quote(log_path)
quoted_pid_path = shlex.quote(pid_path)
quoted_exit_path = shlex.quote(exit_path)
prev_output_len = 0 # track delta for watch pattern scanning
while not session.exited:
time.sleep(2) # Poll every 2 seconds
try:
@@ -420,10 +514,15 @@ class ProcessRegistry:
result = env.execute(f"cat {quoted_log_path} 2>/dev/null", timeout=10)
new_output = result.get("output", "")
if new_output:
# Compute delta for watch pattern scanning
delta = new_output[prev_output_len:] if len(new_output) > prev_output_len else ""
prev_output_len = len(new_output)
with session._lock:
session.output_buffer = new_output
if len(session.output_buffer) > session.max_output_chars:
session.output_buffer = session.output_buffer[-session.max_output_chars:]
if delta:
self._check_watch_patterns(session, delta)
# Check if process is still running
check = env.execute(
@@ -467,6 +566,7 @@ class ProcessRegistry:
session.output_buffer += text
if len(session.output_buffer) > session.max_output_chars:
session.output_buffer = session.output_buffer[-session.max_output_chars:]
self._check_watch_patterns(session, text)
except EOFError:
break
except Exception:
@@ -502,6 +602,7 @@ class ProcessRegistry:
from tools.ansi_strip import strip_ansi
output_tail = strip_ansi(session.output_buffer[-2000:]) if session.output_buffer else ""
self.completion_queue.put({
"type": "completion",
"session_id": session.id,
"command": session.command,
"exit_code": session.exit_code,
@@ -872,6 +973,7 @@ class ProcessRegistry:
"watcher_thread_id": s.watcher_thread_id,
"watcher_interval": s.watcher_interval,
"notify_on_complete": s.notify_on_complete,
"watch_patterns": s.watch_patterns,
})
# Atomic write to avoid corruption on crash
@@ -932,6 +1034,7 @@ class ProcessRegistry:
watcher_thread_id=entry.get("watcher_thread_id", ""),
watcher_interval=entry.get("watcher_interval", 0),
notify_on_complete=entry.get("notify_on_complete", False),
watch_patterns=entry.get("watch_patterns", []),
)
with self._lock:
self._running[session.id] = session

15
tools/terminal_tool.py
View File

@@ -42,7 +42,7 @@ import atexit
import shutil
import subprocess
from pathlib import Path
from typing import Optional, Dict, Any
from typing import Optional, Dict, Any, List
logger = logging.getLogger(__name__)
@@ -1140,6 +1140,7 @@ def terminal_tool(
check_interval: Optional[int] = None,
pty: bool = False,
notify_on_complete: bool = False,
watch_patterns: Optional[List[str]] = None,
) -> str:
"""
Execute a command in the configured terminal environment.
@@ -1154,6 +1155,7 @@ def terminal_tool(
check_interval: Seconds between auto-checks for background processes (gateway only, min 30)
pty: If True, use pseudo-terminal for interactive CLI tools (local backend only)
notify_on_complete: If True and background=True, auto-notify the agent when the process exits
watch_patterns: List of strings to watch for in background output; triggers notification on match
Returns:
str: JSON string with output, exit_code, and error fields
@@ -1439,6 +1441,11 @@ def terminal_tool(
"notify_on_complete": True,
})
# Set watch patterns for output monitoring
if watch_patterns and background:
proc_session.watch_patterns = list(watch_patterns)
result_data["watch_patterns"] = proc_session.watch_patterns
# Register check_interval watcher (gateway picks this up after agent run)
if check_interval and background:
effective_interval = max(30, check_interval)
@@ -1762,6 +1769,11 @@ TERMINAL_SCHEMA = {
"type": "boolean",
"description": "When true (and background=true), you'll be automatically notified when the process finishes — no polling needed. Use this for tasks that take a while (tests, builds, deployments) so you can keep working on other things in the meantime.",
"default": False
},
"watch_patterns": {
"type": "array",
"items": {"type": "string"},
"description": "List of strings to watch for in background process output. When any pattern matches a line of output, you'll be notified with the matching text — like notify_on_complete but triggers mid-process on specific output. Use for monitoring logs, watching for errors, or waiting for specific events (e.g. [\"ERROR\", \"FAIL\", \"listening on port\"])."
}
},
"required": ["command"]
@@ -1779,6 +1791,7 @@ def _handle_terminal(args, **kw):
check_interval=args.get("check_interval"),
pty=args.get("pty", False),
notify_on_complete=args.get("notify_on_complete", False),
watch_patterns=args.get("watch_patterns"),
)