diff --git a/tests/tools/test_completed_process_results.py b/tests/tools/test_completed_process_results.py new file mode 100644 index 0000000000..4048d8c069 --- /dev/null +++ b/tests/tools/test_completed_process_results.py @@ -0,0 +1,181 @@ +"""Completed work remains retrievable when its finite CLI owner exits.""" + +import http.server +import json +import os +from pathlib import Path +import shlex +import subprocess +import sys +import textwrap +import threading + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def test_headless_terminal_result_survives_cli_exit(tmp_path): + """Real CLI, tool dispatch, shell child and fresh reader; only the LLM is local.""" + home = tmp_path / "profile" + home.mkdir() + (home / "config.yaml").write_text( + "model:\n provider: custom\n api_mode: chat_completions\n" + "terminal:\n env: local\n oneshot_completion_wait_seconds: 10\n" + "memory:\n memory_enabled: false\n user_profile_enabled: false\n", + encoding="utf-8", + ) + release = tmp_path / "release" + child = tmp_path / "review.py" + child.write_text(textwrap.dedent(''' + import pathlib, sys, time + deadline = time.monotonic() + 15 + while not pathlib.Path(sys.argv[1]).exists(): + if time.monotonic() > deadline: + sys.exit(91) + time.sleep(0.02) + print("SYNTHETIC_REVIEW_COMPLETE") + print("review stderr", file=sys.stderr) + sys.exit(7) + '''), encoding="utf-8") + argv = [sys.executable, str(child), str(release)] + command = subprocess.list2cmdline(argv) if os.name == "nt" else shlex.join(argv) + observed = [] + + class Provider(http.server.BaseHTTPRequestHandler): + def do_GET(self): + self.send_error(404) + + def do_POST(self): + request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + if "messages" not in request: + self.send_error(404) + return + tool_results = [m for m in request["messages"] if m["role"] == "tool"] + has_terminal = any(t.get("function", {}).get("name") == "terminal" + for t in request.get("tools", [])) + message = {"role": "assistant", "content": "Coordinator finished."} + if has_terminal and not tool_results: + message.update(content=None, tool_calls=[{ + "id": "call_review", "type": "function", "function": { + "name": "terminal", "arguments": json.dumps({ + "command": command, "background": True, "notify": True, + }), + }, + }]) + elif tool_results: + observed.extend(json.loads(m["content"]) for m in tool_results) + release.touch() + response = { + "id": "chatcmpl-local", "object": "chat.completion", "created": 1, + "model": "test-model", "choices": [{ + "index": 0, "message": message, + "finish_reason": "tool_calls" if "tool_calls" in message else "stop", + }], "usage": {"prompt_tokens": 10, "completion_tokens": 10, "total_tokens": 20}, + } + content_type = "application/json" + if request.get("stream"): + response["object"] = "chat.completion.chunk" + response["choices"][0]["delta"] = response["choices"][0].pop("message") + for index, tool in enumerate(message.get("tool_calls", [])): + tool["index"] = index + raw = ("data: " + json.dumps(response) + "\n\ndata: [DONE]\n\n").encode() + content_type = "text/event-stream" + else: + raw = json.dumps(response).encode() + self.send_response(200) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def log_message(self, *args): + pass + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Provider) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + url = f"http://127.0.0.1:{server.server_port}/v1" + env = {**os.environ, "HERMES_HOME": str(home), "HOME": str(tmp_path), + "USERPROFILE": str(tmp_path), "TERMINAL_CWD": str(tmp_path), + "OPENAI_BASE_URL": url, "OPENAI_API_KEY": "local-test-only", + "PYTHONPATH": str(REPO_ROOT)} + try: + producer = subprocess.run([ + sys.executable, "-c", + "import cli; cli.main(query='Run the background review', quiet=True, " + "oneshot=True, provider='custom', model='test-model', api_key='local-test-only', " + f"base_url={url!r}, toolsets='terminal', max_turns=3, ignore_rules=True)", + ], cwd=tmp_path, env=env, capture_output=True, text=True, timeout=60) + finally: + release.touch() + server.shutdown() + server.server_close() + thread.join(timeout=5) + assert producer.returncode == 0, producer.stdout + producer.stderr + assert "Coordinator finished." in producer.stdout + assert len(observed) == 1, (observed, producer.stdout, producer.stderr) + process_id = observed[0]["session_id"] + assert observed[0].get("notify_on_complete") is True, observed + + consumer = textwrap.dedent(''' + import json, sys + import tools.process_registry as pr + from tools.registry import registry + result = registry.get_entry("process_manage").handler( + {"action": "log", "session_id": sys.argv[1]}) + status = registry.get_entry("process_manage").handler( + {"action": "poll", "session_id": sys.argv[1]}) + print(json.dumps({"result": json.loads(result), "status": json.loads(status), + "replayed": not pr.process_registry.completion_queue.empty()})) + ''') + def read_result(profile): + result = subprocess.run([sys.executable, "-c", consumer, process_id], + cwd=tmp_path, env={**env, "HERMES_HOME": str(profile)}, + check=True, capture_output=True, text=True, timeout=30) + return json.loads(result.stdout) + + recovered = read_result(home) + assert recovered["result"]["status"] == "exited", recovered + assert recovered["status"]["exit_code"] == 7 + assert "SYNTHETIC_REVIEW_COMPLETE" in recovered["result"]["output"] + assert "review stderr" in recovered["result"]["output"] + assert recovered["replayed"] is False + assert read_result(tmp_path / "other-profile")["result"]["status"] == "not_found" + + +def test_receipts_are_bounded_redacted_and_session_scoped(tmp_path, monkeypatch): + import time + from tools import process_registry_results as receipts + from tools.process_registry import MAX_OUTPUT_CHARS, ProcessRegistry, ProcessSession + + monkeypatch.setattr(receipts, "MAX_RETAINED_RESULTS", 2) + secret = "sk-" + "aB2cD3eF4gH5iJ6kL7mN8pQ9rS0tU1vW2xY3zA4bC5dE6fG7" + sessions = [] + registry = ProcessRegistry() + for index in range(3): + session = ProcessSession( + id=f"proc_{index:012x}", command=f"echo {secret}", task_id=f"owner-{index}", + owner_task_id=f"owner-{index}", session_key=f"chat-{index}", + started_at=time.time() - receipts.RESULT_RETENTION_SECONDS * 2, + output_buffer="x" * MAX_OUTPUT_CHARS + "\n" + secret, + exited=True, exit_code=index, + ) + registry._running[session.id] = session + registry._move_to_finished(session) + sessions.append(session) + from hermes_constants import get_hermes_home + paths = list((get_hermes_home() / "logs" / "process-results").glob("*.json")) + assert len(paths) == 2 + assert all(secret not in path.read_text(encoding="utf-8") for path in paths) + fresh = ProcessRegistry() + assert fresh.get(sessions[0].id) is None + recovered = fresh.get(sessions[-1].id) + assert recovered.owner_task_id == sessions[-1].owner_task_id + assert len(recovered.output_buffer) <= MAX_OUTPUT_CHARS + assert [s["session_id"] for s in fresh.list_sessions(task_id="owner-2")] == [recovered.id] + assert fresh.list_sessions(task_id="unrelated", session_key="unrelated") == [] + assert fresh.get("proc_0000") is None # Ambiguous across durable results. + assert fresh.completion_queue.empty() + for path in paths: + expired = time.time() - receipts.RESULT_RETENTION_SECONDS - 1 + os.utime(path, (expired, expired)) + assert fresh.get(recovered.id) is None diff --git a/tests/tools/test_oneshot_completion_linger.py b/tests/tools/test_oneshot_completion_linger.py index 79b2c0711c..e9c9fb5b54 100644 --- a/tests/tools/test_oneshot_completion_linger.py +++ b/tests/tools/test_oneshot_completion_linger.py @@ -81,6 +81,7 @@ def test_non_notify_background_processes_are_not_waited_on(registry): def test_already_exited_session_not_waited_on(registry): s = _make_session(exited=True) + s._completion_event.set() with registry._lock: registry._running[s.id] = s result = registry.wait_for_pending_completions(timeout=30) diff --git a/tools/process_registry.py b/tools/process_registry.py index 5305aa0a90..af8de52f29 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -30,9 +30,9 @@ from typing import Any, Dict, List, Optional from hermes_cli.config import get_hermes_home -from agent.redact import redact_sensitive_text from tools.process_registry_notifications import format_process_notification from tools.process_registry_checkpoint import ProcessCheckpointMixin +from tools.process_registry_results import load_completed_results, save_completed_result logger = logging.getLogger(__name__) @@ -785,13 +785,13 @@ class ProcessRegistry(ProcessCheckpointMixin): 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.""" + """Register before starting the reader: an already-exited child can finish immediately.""" reader = threading.Thread(target=reader_target, args=(session, *extra_args), daemon=True, name=reader_name) session._reader_thread = reader - reader.start() with self._lock: self._prune_if_needed() self._running[session.id] = session + reader.start() self._write_checkpoint() def _spawn_local_pty(self, session: ProcessSession, safe_command: str, env_vars: dict) -> ProcessSession: @@ -1172,9 +1172,13 @@ class ProcessRegistry(ProcessCheckpointMixin): Idempotent: kill_process() and the reader thread can both call this; only the FIRST move enqueues the completion notification, so no duplicates.""" with self._lock: - was_running = self._running.pop(session.id, None) is not None + was_running = session.id in self._running + if was_running: + # Keep the session tracked until its result is durable. A finite + # parent must not observe completion and exit during this write. + save_completed_result(session) + self._running.pop(session.id) self._finished[session.id] = session - session._completion_event.set() self._write_checkpoint() if was_running and session.notify_on_complete: notification = { @@ -1192,6 +1196,7 @@ class ProcessRegistry(ProcessCheckpointMixin): } _redact_process_result(notification) self.completion_queue.put(notification) + session._completion_event.set() @staticmethod def _exit_fields(session: ProcessSession) -> dict: @@ -1246,7 +1251,7 @@ class ProcessRegistry(ProcessCheckpointMixin): with self._lock: pending = [ s for s in self._running.values() - if s.notify_on_complete and not s.exited and (task_id is None or s.task_id == task_id) + if s.notify_on_complete and not s._completion_event.is_set() and (task_id is None or s.task_id == task_id) ] if not pending or timeout <= 0: return result @@ -1264,7 +1269,7 @@ class ProcessRegistry(ProcessCheckpointMixin): interrupted = False for session in pending: try: - while not session.exited: + while not session._completion_event.is_set(): if interrupted or _is_interrupted(): interrupted = True break @@ -1283,14 +1288,14 @@ class ProcessRegistry(ProcessCheckpointMixin): # #17327). self._reconcile_local_exit(session) self._refresh_detached_session(session) - if session.exited: + if session._completion_event.is_set(): break session._completion_event.wait(min(remaining, interval)) except KeyboardInterrupt: # Stop waiting, but never let the interrupt skip the caller's durable # teardown (session flush, end_session) that follows. interrupted = True - result["completed" if session.exited else "timed_out"].append(session.id) + result["completed" if session._completion_event.is_set() else "timed_out"].append(session.id) if result["timed_out"]: logger.warning( "One-shot exit linger timed out after %ss with %d background " @@ -1400,8 +1405,12 @@ class ProcessRegistry(ProcessCheckpointMixin): def get(self, session_id: str) -> Optional[ProcessSession]: """Session by full ID or unique prefix (``proc_4dae`` / bare ``4dae``, like git short hashes); ambiguous or too-short prefixes resolve to None, never a guess.""" + if not isinstance(session_id, str) or not session_id: + return None with self._lock: session = self._running.get(session_id) or self._finished.get(session_id) + if session is None: + session = load_completed_results(session_id).get(session_id) return self._refresh_detached_session(session if session is not None else self._resolve_prefix(session_id)) def _resolve_prefix(self, session_id: str) -> Optional[ProcessSession]: @@ -1414,12 +1423,13 @@ class ProcessRegistry(ProcessCheckpointMixin): query = f"proc_{query}" if len(query) - len("proc_") < self._MIN_PREFIX_CHARS: return None + matches = load_completed_results(query) with self._lock: - matches = [ - s for store in (self._running, self._finished) + matches.update({ + sid: s for store in (self._running, self._finished) for sid, s in store.items() if sid.startswith(query) - ] - return matches[0] if len(matches) == 1 else None + }) + return next(iter(matches.values())) if len(matches) == 1 else None def _reconcile_local_exit(self, session: "ProcessSession") -> None: """Reconcile ``session.exited`` against the real child state. @@ -1781,9 +1791,11 @@ class ProcessRegistry(ProcessCheckpointMixin): surfaced too, even if they belong to a different task — so the agent can discover a forgotten preview server that is blocking session reset (#29177). """ + sessions = load_completed_results() with self._lock: - all_sessions = list(self._running.values()) + list(self._finished.values()) - all_sessions = [self._refresh_detached_session(s) for s in all_sessions] + sessions.update(self._finished) + sessions.update(self._running) + all_sessions = [self._refresh_detached_session(s) for s in sessions.values()] if task_id or session_key: all_sessions = [ s for s in all_sessions @@ -1910,6 +1922,8 @@ PROCESS_SCHEMA = { "description": ( "Poll, wait on, or kill background terminal processes (from " "terminal(background=true)). " + "Completed results remain retrievable by session_id after restart " + "(up to 7 days, newest 64 results per profile; rolling output tail). " "poll: status + new output. log: full output, paged. wait: block " "until exit or timeout (partial output on timeout). write vs " "submit: submit appends Enter — use it to answer prompts; write " diff --git a/tools/process_registry_results.py b/tools/process_registry_results.py new file mode 100644 index 0000000000..39279479d0 --- /dev/null +++ b/tools/process_registry_results.py @@ -0,0 +1,89 @@ +"""Bounded, profile-local receipts for completed terminal processes. + +Receipts are read through process_manage, never replayed as notifications or +adopted as live PIDs. Each producer writes its own file so independent one-shot +parents cannot overwrite each other's results in the running-PID checkpoint. +""" + +import json +import logging +import re +import time + +from hermes_constants import get_hermes_home +from utils import atomic_json_write + +logger = logging.getLogger("tools.process_registry") + +RESULT_RETENTION_SECONDS = 7 * 24 * 60 * 60 +MAX_RETAINED_RESULTS = 64 +_RESULT_FIELDS = ( + "id", "command", "cwd", "task_id", "owner_task_id", "session_key", + "parent_session_id", "started_at", "exit_code", "completion_reason", + "termination_source", "notify_on_complete", +) + + +def _result_paths(): + """Prune by completion time, not start time (jobs can take days).""" + directory = get_hermes_home() / "logs" / "process-results" + cutoff = time.time() - RESULT_RETENTION_SECONDS + retained = [] + for path in directory.glob("proc_*.json"): + try: + modified = path.stat().st_mtime + if modified < cutoff: + path.unlink(missing_ok=True) + else: + retained.append((modified, path)) + except FileNotFoundError: + continue # Another producer pruned it. + retained.sort(key=lambda item: (item[0], item[1].name), reverse=True) + for _, path in retained[MAX_RETAINED_RESULTS:]: + path.unlink(missing_ok=True) + return [path for _, path in retained[:MAX_RETAINED_RESULTS]] + + +def save_completed_result(session) -> None: + from tools.process_registry import MAX_OUTPUT_CHARS, _redact_process_result + + with session._lock: + record = {key: getattr(session, key) for key in _RESULT_FIELDS} + record["output"] = session.output_buffer[-MAX_OUTPUT_CHARS:] + _redact_process_result(record) + directory = get_hermes_home() / "logs" / "process-results" + try: + directory.mkdir(mode=0o700, parents=True, exist_ok=True) + atomic_json_write(directory / f"{session.id}.json", record, mode=0o600) + _result_paths() + except OSError: + # Preserve live delivery on disk failure, but never silently claim durability. + logger.warning("Could not retain completed process result %s", session.id, exc_info=True) + + +def load_completed_results(prefix: str = "") -> dict: + """Restore read-only snapshots; no process handles, watchers, or queue events.""" + from tools.process_registry import ProcessSession + + results = {} + try: + paths = _result_paths() + except OSError: + logger.warning("Could not read retained process results", exc_info=True) + return results + for path in paths: + if not path.stem.startswith(prefix): + continue + try: + record = json.loads(path.read_text(encoding="utf-8")) + if record["id"] != path.stem or not re.fullmatch(r"proc_[\w]+", record["id"]): + continue + session = ProcessSession( + **{key: record[key] for key in _RESULT_FIELDS}, + exited=True, output_buffer=record["output"], + ) + session._completion_event.set() + results[session.id] = session + except (OSError, ValueError, KeyError, TypeError): + logger.debug("Skipping unreadable process result %s", path.name, exc_info=True) + return results diff --git a/website/docs/developer-guide/tools-runtime.md b/website/docs/developer-guide/tools-runtime.md index c863219eb2..b030f88ae0 100644 --- a/website/docs/developer-guide/tools-runtime.md +++ b/website/docs/developer-guide/tools-runtime.md @@ -224,6 +224,15 @@ It also supports: - PTY mode - approval callbacks for dangerous commands +`tools/process_registry_checkpoint.py` owns running-process checkpoints and +PID-safe adoption. Completed output is separate: `tools/process_registry_results.py` +writes one atomic, redacted receipt per process under the profile's +`logs/process-results/`. Producers cannot overwrite another parent's results by +rewriting the shared PID checkpoint. The registry persists the receipt before +releasing its completion event; one-shot linger waits on that event. The existing +process query methods load retained snapshots without adopting PIDs or enqueuing +notifications. Receipt retention is bounded by age and count. + ## Concurrency Tool calls may execute sequentially or concurrently depending on the tool mix and interaction requirements. diff --git a/website/docs/user-guide/features/tools.md b/website/docs/user-guide/features/tools.md index ebf2adc7ce..d3059746f9 100644 --- a/website/docs/user-guide/features/tools.md +++ b/website/docs/user-guide/features/tools.md @@ -231,6 +231,20 @@ process(action="write", session_id="proc_abc123", data="y") # Send input PTY mode (`pty=true`) enables interactive CLI tools like Codex and Claude Code. +Completed background commands retain their exit status and captured output in the +active profile. After a headless parent exits or Hermes restarts, use the original +`session_id` with `process(action="log")` for output and `process(action="poll")` +for exit status. `process(action="list")` also includes retained results for the +current task or conversation. + +Hermes keeps the newest **64 completed results**, for up to **7 days after +completion**, under `logs/process-results/` in the profile's Hermes home. Each +receipt contains at most the existing rolling **200,000-character output tail**, +with the same secret redaction as terminal output. Receipts expire on subsequent +result reads or writes. Recovery does not rerun commands or replay completion +notifications. This preserves work that finished while the parent was alive; +it does not keep unfinished children alive after a timeout or crash. + ## Sudo Support On an interactive parent session, supported sudo commands use the masked password prompt (cached for the session). This includes literal absolute or quoted executable paths and `env` prefixes with ordinary options and assignments, such as `env -u UNUSED /usr/bin/sudo id`. Passwordless sudo does not need a prompt. You can also configure `SUDO_PASSWORD` in your profile's `.env` file on the agent machine.