fix(tools): retain completed background process results across exit
This commit is contained in:
maximilliangrand
committed by
Teknium
Teknium
parent
5695ebc40f
commit
b72e373e23
@@ -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
|
||||
@@ -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)
|
||||
|
||||
+29
-15
@@ -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 "
|
||||
|
||||
@@ -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
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user