ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(tools): process_registry — _finish_reader/_owns_event/_status_head/_spawn_env helpers; notifications list building compacted

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:23:25 -07:00
parent 870aea6e47
commit 2c2935c4c8
2 changed files with 118 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/process_registry.py
+99 -123
View File
@@ -434,12 +434,11 @@ class ProcessRegistry:
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
pat = next((p for p in session.watch_patterns if p in line), None) # one match per line
if pat is not None:
matched_lines.append(line.rstrip())
if matched_pattern is None:
matched_pattern = pat
if not matched_lines:
return
@@ -767,8 +766,9 @@ class ProcessRegistry:
return "/tmp"
def _scope_argv(self, session: ProcessSession, safe_command: str, unit_suffix: str, label: str) -> List[str]:
"""Login-shell argv for *safe_command*, wrapped in a transient systemd scope when
we are the supervised gateway (own cgroup: an OOM kills only the worker, not the
"""Login-shell argv for *safe_command* (parity with LocalEnvironment: rc files
sourced, user tools on PATH), wrapped in a transient systemd scope when we are
the supervised gateway (own cgroup: an OOM kills only the worker, not the
gateway and its messaging control plane)."""
argv = [_find_shell(), "-lic", f"set +m; {safe_command}"]
in_supervised_gateway = _IS_LINUX and _is_supervised_gateway_process()
@@ -780,14 +780,21 @@ class ProcessRegistry:
# the whole gateway down.
logger.debug(
"%s background executor not isolated in a systemd scope "
"(systemd-run --user unavailable); worker shares the gateway cgroup.",
label,
"(systemd-run --user unavailable); worker shares the gateway cgroup.", label,
)
return argv
def _track_started(self, session: ProcessSession, reader_target, reader_name: str) -> None:
@staticmethod
def _spawn_env(env_vars: dict) -> dict:
"""Sanitized child env; PYTHONUNBUFFERED so tqdm/datasets-style buffering
doesn't hide progress from process(action="poll")."""
env = _sanitize_subprocess_env(os.environ, env_vars)
env["PYTHONUNBUFFERED"] = "1"
return env
def _track_started(self, session: ProcessSession, reader_target, reader_name: str, extra_args=()) -> None:
"""Start the output reader thread, register the session and checkpoint it."""
reader = threading.Thread(target=reader_target, args=(session,), daemon=True, name=reader_name)
reader = threading.Thread(target=reader_target, args=(session, *extra_args), daemon=True, name=reader_name)
session._reader_thread = reader
reader.start()
with self._lock:
@@ -805,8 +812,7 @@ class ProcessRegistry:
from winpty import PtyProcess as _PtyProcessCls
else:
from ptyprocess import PtyProcess as _PtyProcessCls
pty_env = _sanitize_subprocess_env(os.environ, env_vars)
pty_env["PYTHONUNBUFFERED"] = "1"
pty_env = self._spawn_env(env_vars)
# A PTY is a real TTY, so pager-happy tools (git log/diff, man) WILL page and
# hang waiting for `q` — default them to cat, honoring any pager the user set.
pty_env.setdefault("GIT_PAGER", "cat")
@@ -851,11 +857,7 @@ class ProcessRegistry:
) from e
session.systemd_unit = ""
# Pipe path (non-PTY or PTY fallback). The user's login shell keeps parity with
# LocalEnvironment (rc files sourced, user tools on PATH). PYTHONUNBUFFERED so
# tqdm/datasets-style buffering doesn't hide progress from process(action="poll").
bg_env = _sanitize_subprocess_env(os.environ, env_vars)
bg_env["PYTHONUNBUFFERED"] = "1"
# Pipe path (non-PTY or PTY fallback).
_popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {}
unit_suffix = f"{session.id}-pipe-fallback" if pty_scope_attempted else session.id
spawn_argv = self._scope_argv(session, safe_command, unit_suffix, "Local")
@@ -865,8 +867,8 @@ class ProcessRegistry:
# session (observed as dead TUIs in state T). Cgroup isolation is unaffected —
# the scope attaches to the invoked process, not the spawning session.
proc = subprocess.Popen(
spawn_argv, text=True, cwd=session.cwd, env=bg_env, encoding="utf-8", errors="replace",
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
spawn_argv, text=True, cwd=session.cwd, env=self._spawn_env(env_vars), encoding="utf-8",
errors="replace", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
start_new_session=True, **_popen_kwargs,
)
session.process = proc
@@ -923,11 +925,7 @@ class ProcessRegistry:
try:
result = env.execute(bg_command, timeout=timeout, rewrite_compound_background=False)
output = result.get("output", "").strip()
for line in output.splitlines():
line = line.strip()
if line.isdigit():
session.pid = int(line)
break
session.pid = next((int(ln) for ln in map(str.strip, output.splitlines()) if ln.isdigit()), None)
# No PID from the wrapper (syntax error, broken redirect): a failed launch,
# not a fake running session.
if session.pid is None:
@@ -936,22 +934,13 @@ class ProcessRegistry:
except Exception as e:
session.mark_exited(-1, "failed_start", "failed_start")
session.output_buffer = f"Failed to start: {e}"
if session.exited:
with self._lock:
self._prune_if_needed()
return session
# Poller thread periodically reads the log file
reader = threading.Thread(
target=self._env_poller_loop, args=(session, env, log_path, pid_path, exit_path),
daemon=True, name=f"proc-poller-{session.id}",
self._track_started(
session, self._env_poller_loop, f"proc-poller-{session.id}", (env, log_path, pid_path, exit_path),
)
session._reader_thread = reader
reader.start()
with self._lock:
self._prune_if_needed()
self._running[session.id] = session
self._write_checkpoint()
return session
# ----- Reader / Poller Threads -----
@@ -997,14 +986,12 @@ class ProcessRegistry:
# select() needs a real OS fd; mocked streams (tests, adapters) may lack
# fileno() and use the blocking read instead.
fd = None
if raw_read is not None and not _IS_WINDOWS:
try:
candidate = stdout.fileno()
except Exception:
candidate = None
if isinstance(candidate, int) and candidate >= 0:
fd = candidate
try:
fd = stdout.fileno() if raw_read is not None and not _IS_WINDOWS else None
except Exception:
fd = None
if not (isinstance(fd, int) and fd >= 0):
fd = None
if fd is not None:
import select as _select
idle_after_exit = 0
@@ -1032,20 +1019,25 @@ class ProcessRegistry:
except Exception as e:
logger.debug("Process stdout reader ended: %s", e)
finally:
# Flush the decoder: a truncated multibyte sequence at EOF becomes one
# U+FFFD instead of vanishing.
try:
tail = decoder.decode(b"", final=True)
if tail:
_append_chunk(tail)
except Exception:
pass
# Always reap the child to prevent zombie processes.
try:
session.process.wait(timeout=5)
except Exception as e:
logger.debug("Process wait timed out or failed: %s", e)
self._finish_exited(session, session.process.returncode)
self._finish_reader(
session, decoder, _append_chunk, "Process",
lambda: session.process.wait(timeout=5), lambda: session.process.returncode,
)
def _finish_reader(self, session, decoder, append, label, wait, exit_code) -> None:
"""Reader-thread teardown: flush the decoder (a truncated multibyte tail becomes
one U+FFFD instead of vanishing), reap the child (no zombies), record the exit."""
try:
tail = decoder.decode(b"", final=True)
if tail:
append(tail)
except Exception:
pass
try:
wait()
except Exception as e:
logger.debug("%s wait timed out or failed: %s", label, e)
self._finish_exited(session, exit_code())
def _env_poller_loop(self, session: ProcessSession, env: Any, log_path: str, pid_path: str, exit_path: str):
"""Background thread: poll a sandbox log file for non-local backends."""
@@ -1080,10 +1072,8 @@ class ProcessRegistry:
return
except Exception:
# Environment might be gone (sandbox reaped, etc.)
session.exited = True
session.exit_code = -1
session.completion_reason = "lost"
session.termination_source = "backend_lost"
session.exited, session.exit_code = True, -1
session.completion_reason, session.termination_source = "lost", "backend_lost"
self._move_to_finished(session)
return
@@ -1105,18 +1095,10 @@ class ProcessRegistry:
break
except Exception as e:
logger.debug("PTY stdout reader ended: %s", e)
# Flush any partial multibyte sequence held by the decoder.
try:
tail = decoder.decode(b"", final=True)
if tail:
self._ingest_output(session, tail)
except Exception:
pass
try:
pty.wait()
except Exception as e:
logger.debug("PTY wait timed out or failed: %s", e)
self._finish_exited(session, pty.exitstatus if hasattr(pty, 'exitstatus') else -1)
self._finish_reader(
session, decoder, lambda t: self._ingest_output(session, t), "PTY",
pty.wait, lambda: pty.exitstatus if hasattr(pty, 'exitstatus') else -1,
)
def _ingest_output(self, session: ProcessSession, text: str) -> None:
"""Buffer a freshly-read chunk, then scan watch patterns and stream it live."""
@@ -1280,6 +1262,22 @@ class ProcessRegistry:
except Exception:
return False
@staticmethod
def _owns_event(evt: dict, session_key: str, owns_event, is_async_delegation: bool) -> bool:
"""Routing verdict for one drained event (see drain_notifications); False = requeue."""
evt_session_key = str(evt.get("session_key") or "")
requires_positive_proof = is_async_delegation or bool(evt_session_key or evt.get("origin_ui_session_id"))
if owns_event is not None and requires_positive_proof:
try:
return bool(owns_event(evt))
except Exception:
return False # fail closed — never leak on a broken check
if session_key and requires_positive_proof:
return evt_session_key == session_key
# Restored payloads from a previous process: an unfiltered drain cannot prove
# ownership, so leave them for the owner.
return not (is_async_delegation and evt.get("restored"))
def drain_notifications(
self, session_key: str = "", owns_event=None, *, skip_poll_observed: bool = True,
) -> "list[tuple[dict, str]]":
@@ -1305,25 +1303,7 @@ class ProcessRegistry:
except Exception:
break
is_async_delegation = evt.get("type") == "async_delegation"
evt_session_key = str(evt.get("session_key") or "")
requires_positive_proof = is_async_delegation or bool(
evt_session_key or evt.get("origin_ui_session_id")
)
if owns_event is not None and requires_positive_proof:
try:
owned = bool(owns_event(evt))
except Exception:
owned = False # fail closed — never leak on a broken check
if not owned:
requeue.append(evt)
continue
elif session_key and requires_positive_proof:
if evt_session_key != session_key:
requeue.append(evt)
continue
elif is_async_delegation and evt.get("restored"):
# Restored payloads from a previous process: an unfiltered drain
# cannot prove ownership, so leave them for the owner.
if not self._owns_event(evt, session_key, owns_event, is_async_delegation):
requeue.append(evt)
continue
# Routing happened first so a foreign session cannot drop the owner's
@@ -1438,6 +1418,14 @@ class ProcessRegistry:
)
self._move_to_finished(session)
@staticmethod
def _status_head(session: ProcessSession) -> dict:
return {
"session_id": session.id,
"command": session.command,
"status": "exited" if session.exited else "running",
}
def poll(self, session_id: str) -> dict:
"""Check status and get new output for a background process."""
session = self.get(session_id)
@@ -1447,9 +1435,7 @@ class ProcessRegistry:
with session._lock:
output_preview = _output_tail(session, 1000)
result = {
"session_id": session.id,
"command": session.command,
"status": "exited" if session.exited else "running",
**self._status_head(session),
"pid": session.pid,
"uptime_seconds": int(time.time() - session.started_at),
"output_preview": output_preview,
@@ -1489,9 +1475,7 @@ class ProcessRegistry:
stop = slice(offset, offset + limit).indices(total_lines)[1]
observed_completion_output = total_lines == 0 or (bool(selected) and stop == total_lines)
result = {
"session_id": session.id,
"command": session.command,
"status": "exited" if session.exited else "running",
**self._status_head(session),
"output": "\n".join(selected),
"total_lines": total_lines,
"showing": f"{len(selected)} lines",
@@ -1532,6 +1516,7 @@ class ProcessRegistry:
if session is None:
return _not_found(session_id)
self._reconcile_local_exit(session) # orphaned-pipe reader guard
result = None
if session.exited:
self._completion_consumed.add(session_id)
result = self._exit_snapshot(session, "exited")
@@ -1542,8 +1527,6 @@ class ProcessRegistry:
"output": _output_tail(session, 1000),
"note": "User sent a new message -- wait interrupted",
}
else:
result = None
if result is not None:
if timeout_note:
result["timeout_note"] = timeout_note
@@ -1564,13 +1547,12 @@ class ProcessRegistry:
base_note = f"Wait window of {effective_timeout}s elapsed — the process is still running. This is not an error."
if session.started_at:
base_note += f" Uptime: {int(time.time() - session.started_at)}s."
if session.notify_on_complete:
base_note += " notify_on_complete is set: you will be notified on exit — do more work instead of waiting again."
else:
base_note += (
" Poll again later or use terminal(background=true, "
"notify_on_complete=true) next time for automatic notification."
)
base_note += (
" notify_on_complete is set: you will be notified on exit — do more work instead of waiting again."
if session.notify_on_complete else
" Poll again later or use terminal(background=true, "
"notify_on_complete=true) next time for automatic notification."
)
result["timeout_note"] = f"{timeout_note}. {base_note}" if timeout_note else base_note
return result
@@ -1676,10 +1658,8 @@ class ProcessRegistry:
else:
return {
"status": "error",
"error": (
"Recovered process cannot be killed after restart because "
"its original runtime handle is no longer available"
),
"error": "Recovered process cannot be killed after restart because "
"its original runtime handle is no longer available",
}
return None
@@ -1724,20 +1704,18 @@ class ProcessRegistry:
"""Send data + newline to stdin (like pressing Enter).
On a Windows PTY, Enter is a carriage return: ConPTY treats ``\\r`` as
end-of-line and a bare ``\\n`` through pywinpty is NOT a line terminator —
the child's blocking line read (``readline()``, Go ``bufio.Scanner`` in
``gh auth login``) never returns and the process hangs looking healthy.
``\\r\\n`` gives it both; POSIX PTYs and pipes keep ``\\n``.
end-of-line and a bare ``\\n`` through pywinpty is NOT a line terminator — the
child's blocking line read (``readline()``, Go ``bufio.Scanner``) never returns
and the process hangs looking healthy. ``\\r\\n`` gives it both; POSIX keeps ``\\n``.
"""
session = self.get(session_id)
is_windows_pty = bool(_IS_WINDOWS and session is not None and session._pty)
return self.write_stdin(session_id, data + ("\r\n" if is_windows_pty else "\n"))
def request_close_terminal(self, session_id: str) -> dict:
"""Ask the desktop GUI to close this process's read-only terminal tab.
Does NOT kill the process — output keeps buffering and the tab can be
reopened from the status stack. Errors when no UI close sink is wired."""
"""Ask the desktop GUI to close this process's read-only terminal tab. Does NOT
kill the process — output keeps buffering and the tab can be reopened from the
status stack. Errors when no UI close sink is wired."""
sink = self.on_close
if sink is None:
return {"status": "error", "error": "close_terminal is only available in the Hermes desktop app."}
@@ -1882,7 +1860,6 @@ class ProcessRegistry:
now = time.time()
expired = [sid for sid, s in self._finished.items() if (now - s.started_at) > FINISHED_TTL_SECONDS]
if len(self._running) + len(self._finished) - len(expired) >= MAX_PROCESSES:
# Still over the limit: also drop the oldest surviving finished session.
survivors = [sid for sid in self._finished if sid not in expired]
if survivors:
expired.append(min(survivors, key=lambda sid: self._finished[sid].started_at))
@@ -2114,7 +2091,6 @@ def _handle_process(args, **kw):
action = args.get("action", "")
# Coerce to string — some models send session_id as an integer
session_id = str(args.get("session_id", "")) if args.get("session_id") is not None else ""
if action == "list":
return json.dumps(_list_processes(kw.get("task_id")), ensure_ascii=False)
if action in _SESSION_ACTIONS:
tools/process_registry_notifications.py
+19 -36
View File
@@ -78,8 +78,7 @@ def _delegation_model_not_found_notice(results) -> "list[str] | None":
f'"{model}" was rejected by provider "{provider}" '
"(HTTP 400: not a valid model ID).",
"Every task in this batch failed for this reason before doing any work.",
"Check Settings → Advanced → Subagent Model (or: "
"hermes config get delegation.model).",
"Check Settings → Advanced → Subagent Model (or: hermes config get delegation.model).",
]
try:
from hermes_cli.fallback_config import get_fallback_chain
@@ -144,15 +143,13 @@ def _format_batch_delegation(evt: dict, deleg_id: str, completed_at: float) -> s
lines.extend(_task_source_lines(evt))
lines.append(f"{_role_model(evt)} Total duration: {total_dur}s")
if error and not results:
lines.append("--- ERROR ---")
lines.append(f"The batch did not complete successfully: {error}")
lines += ["--- ERROR ---", f"The batch did not complete successfully: {error}"]
return "\n".join(lines)
# Config-level rejection notice BEFORE the per-task wall — a rejected
# delegation model fails every task identically and must not stay buried.
_notice = _delegation_model_not_found_notice(results)
if _notice:
lines.append("")
lines.extend(_notice)
lines += ["", *_notice]
for r in sorted(results, key=lambda x: x.get("task_index", 0)):
idx = r.get("task_index", 0)
r_status = r.get("status", "?")
@@ -161,19 +158,14 @@ def _format_batch_delegation(evt: dict, deleg_id: str, completed_at: float) -> s
r_goal = goals[idx] if idx < len(goals) else r.get("goal", "")
r_truncated = _is_truncated(r)
icon = "⚠" if r_truncated else ("✓" if r_status in ("completed", "success") else "✗")
lines.append("")
header = f"--- {icon} TASK {idx + 1}/{n}"
if r_goal:
header += f": {r_goal}"
header += f" (status={r_status}"
header = f"--- {icon} TASK {idx + 1}/{n}" + (f": {r_goal}" if r_goal else "") + f" (status={r_status}"
if r.get("api_calls"):
header += f", api_calls={r['api_calls']}"
if r.get("duration_seconds") is not None:
header += f", {r['duration_seconds']}s"
if r_truncated:
header += ", TRUNCATED: hit max_iterations — work may be incomplete"
header += ") ---"
lines.append(header)
lines += ["", header + ") ---"]
if r_status in ("completed", "success") and r_summary:
if r_truncated:
lines.append(_TRUNCATED_SUMMARY_NOTE)
@@ -181,30 +173,26 @@ def _format_batch_delegation(evt: dict, deleg_id: str, completed_at: float) -> s
elif r_summary:
if r_error:
lines.append(f"({r_status}: {r_error})")
lines.append("Partial output:")
lines.append(r_summary)
lines += ["Partial output:", r_summary]
else:
lines.append(f"(no summary — status={r_status}" + (f": {r_error}" if r_error else "") + ")")
r_live = r.get("live_transcript")
if r_live:
lines.append(f"Full live transcript (complete tool/assistant trace): {r_live}")
if r.get("live_transcript"):
lines.append(f"Full live transcript (complete tool/assistant trace): {r['live_transcript']}")
return "\n".join(lines)
def _format_async_delegation(evt: dict) -> str:
"""Format an async-delegation completion into a self-contained re-injection.
Carries the FULL original task source (goal, context, toolsets, role,
model) plus dispatch time, status, and the complete result summary: when
this re-enters the conversation the agent may be deep in unrelated context
and must be able to use the result OR re-dispatch without remembering why
the subagent existed.
Carries the FULL original task source (goal, context, toolsets, role, model) plus
dispatch time, status, and the complete result summary: when this re-enters the
conversation the agent may be deep in unrelated context and must be able to use
the result OR re-dispatch without remembering why the subagent existed.
"""
deleg_id = evt.get("delegation_id", "unknown")
completed_at = evt.get("completed_at") or time.time()
if evt.get("is_batch") or isinstance(evt.get("results"), list):
return _format_batch_delegation(evt, deleg_id, completed_at)
status = evt.get("status") or "completed"
summary = evt.get("summary")
error = evt.get("error")
@@ -222,14 +210,13 @@ def _format_async_delegation(evt: dict) -> str:
lines.append(_role_model(evt))
_notice = _delegation_model_not_found_notice([evt])
if _notice:
lines.append("")
lines.extend(_notice)
lines += ["", *_notice]
_trunc = " [TRUNCATED: hit max_iterations — work may be incomplete]" if truncated else ""
lines.append(
lines += [
f"Status: {status} API calls: {evt.get('api_calls', 0)} "
f"Duration: {evt.get('duration_seconds', '?')}s{_trunc}"
)
lines.append("--- RESULT ---")
f"Duration: {evt.get('duration_seconds', '?')}s{_trunc}",
"--- RESULT ---",
]
if status in ("completed", "success") and summary:
if truncated:
lines.append(_TRUNCATED_SUMMARY_NOTE)
@@ -242,8 +229,7 @@ def _format_async_delegation(evt: dict) -> str:
f"The subagent did not complete successfully (status={status})." + (f"\n{error}" if error else "")
)
if summary:
lines.append("Partial output:")
lines.append(summary)
lines += ["Partial output:", summary]
return "\n".join(lines)
@@ -271,10 +257,7 @@ def _delegation_attribution_line(evt: dict) -> "str | None":
if len(goal) > 120:
goal = goal[:117] + "..."
deleg = info.get("delegation_id")
parts = [f"Started by subagent {task_id}"]
if deleg:
parts.append(f"of delegation {deleg}")
line = " ".join(parts) + "."
line = f"Started by subagent {task_id}" + (f" of delegation {deleg}" if deleg else "") + "."
if goal:
line += f' Task: "{goal}"'
return line