fix(tools): keep retained terminal results scoped to their owner
Capture the durable parent session before output readers start, including CLI and non-notifying spawns. Require that parent or its compression continuation for retained reads; exact and prefix handles alone do not authorize access. Live Linux terminal/one-shot linger/fresh-reader A/B: base loses results; updated owner recovers both streams and exit 7. Unbound, foreign session, delegated child, and other profile cannot recover the receipt. No notifications are replayed. Full tools suite is queued behind the campaign test lock. Follow-up to contributor salvage #104805 for #104511.
This commit is contained in:
Teknium
59
evals/process_result_receipt_probe.py
Normal file
59
evals/process_result_receipt_probe.py
Normal file
@@ -0,0 +1,59 @@
|
||||
"""Live process I/O proof, without model calls: python probe.py REPO OUT_DIR."""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
repo, out = map(Path, sys.argv[1:3])
|
||||
out.mkdir(parents=True, exist_ok=True)
|
||||
producer = '''
|
||||
import json, shlex, sys
|
||||
from gateway.session_context import scoped_current_session_id
|
||||
from tools.terminal_tool import terminal_tool
|
||||
from tools.process_registry import process_registry
|
||||
from hermes_cli.oneshot import _linger_for_background_completions
|
||||
with scoped_current_session_id("receipt-owner"):
|
||||
code = "import sys,time; time.sleep(.2); print('RECEIPT_STDOUT'); print('RECEIPT_STDERR',file=sys.stderr); sys.exit(7)"
|
||||
result = json.loads(terminal_tool(shlex.join([sys.executable, '-c', code]), background=True,
|
||||
notify_on_complete=True, task_id='receipt-task'))
|
||||
_linger_for_background_completions()
|
||||
session = process_registry.get(result['session_id'])
|
||||
session._reader_thread.join(timeout=10)
|
||||
print(json.dumps({'spawn': result, 'live': process_registry.read_log(session.id)}))
|
||||
'''
|
||||
consumer = '''
|
||||
import json,sys
|
||||
from gateway.session_context import scoped_current_session_id
|
||||
from tools.process_registry import process_registry
|
||||
with scoped_current_session_id(sys.argv[2]):
|
||||
print(json.dumps({'log':process_registry.read_log(sys.argv[1]),
|
||||
'poll':process_registry.poll(sys.argv[1]),
|
||||
'replayed':process_registry.completion_queue.qsize()}))
|
||||
'''
|
||||
env = {key: value for key, value in os.environ.items()
|
||||
if not key.startswith(('HERMES_', 'OPENAI_', 'ANTHROPIC_', 'TERMINAL_'))
|
||||
and not key.endswith(('_API_KEY', '_TOKEN', '_SECRET'))}
|
||||
env.update(HOME=str(out), HERMES_HOME=str(out / 'profile'), PYTHONPATH=str(repo),
|
||||
TERMINAL_CWD=str(out), PYTHONDONTWRITEBYTECODE='1')
|
||||
|
||||
def run(code, *args, profile=None):
|
||||
child_env = dict(env)
|
||||
if profile:
|
||||
child_env['HERMES_HOME'] = str(out / profile)
|
||||
result = subprocess.run([sys.executable, '-c', code, *args], cwd=repo,
|
||||
env=child_env, stdin=subprocess.DEVNULL,
|
||||
capture_output=True, text=True, encoding='utf-8', timeout=45)
|
||||
if result.returncode:
|
||||
raise RuntimeError(result.stdout + result.stderr)
|
||||
return json.loads(result.stdout.splitlines()[-1])
|
||||
|
||||
before_exit = run(producer)
|
||||
sid = before_exit['spawn']['session_id']
|
||||
result = {'repo': str(repo), 'before_exit': before_exit,
|
||||
'owner': run(consumer, sid, 'receipt-owner'),
|
||||
'stranger': run(consumer, sid, 'receipt-stranger'),
|
||||
'unbound': run(consumer, sid, ''),
|
||||
'other_profile': run(consumer, sid, 'receipt-owner', profile='other-profile')}
|
||||
(out / 'result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
|
||||
print(json.dumps(result, indent=2))
|
||||
@@ -104,7 +104,8 @@ def test_headless_terminal_result_survives_cli_exit(tmp_path):
|
||||
"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)
|
||||
], cwd=tmp_path, env=env, stdin=subprocess.DEVNULL,
|
||||
capture_output=True, text=True, encoding="utf-8", timeout=60)
|
||||
finally:
|
||||
release.touch()
|
||||
server.shutdown()
|
||||
@@ -130,9 +131,12 @@ def test_headless_terminal_result_survives_cli_exit(tmp_path):
|
||||
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)
|
||||
check=True, stdin=subprocess.DEVNULL, capture_output=True,
|
||||
text=True, encoding="utf-8", timeout=30)
|
||||
return json.loads(result.stdout)
|
||||
|
||||
receipt = json.loads((home / "logs" / "process-results" / f"{process_id}.json").read_text(encoding="utf-8"))
|
||||
env["HERMES_SESSION_ID"] = receipt["parent_session_id"]
|
||||
recovered = read_result(home)
|
||||
assert recovered["result"]["status"] == "exited", recovered
|
||||
assert recovered["status"]["exit_code"] == 7, recovered
|
||||
@@ -149,12 +153,16 @@ def test_receipts_are_bounded_redacted_and_session_scoped(tmp_path, monkeypatch)
|
||||
|
||||
monkeypatch.setattr(receipts, "MAX_RETAINED_RESULTS", 2)
|
||||
secret = "sk-" + "aB2cD3eF4gH5iJ6kL7mN8pQ9rS0tU1vW2xY3zA4bC5dE6fG7"
|
||||
from gateway.session_context import scoped_current_session_id
|
||||
from tools.process_registry_results import load_completed_results
|
||||
monkeypatch.setenv("HERMES_SESSION_ID", "owner-session")
|
||||
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}",
|
||||
parent_session_id="owner-session",
|
||||
started_at=time.time() - receipts.RESULT_RETENTION_SECONDS * 2,
|
||||
output_buffer="x" * MAX_OUTPUT_CHARS + "\n" + secret,
|
||||
exited=True, exit_code=index,
|
||||
@@ -176,6 +184,22 @@ def test_receipts_are_bounded_redacted_and_session_scoped(tmp_path, monkeypatch)
|
||||
task_id="owner-2", include_retained=True)] == [recovered.id]
|
||||
assert fresh.list_sessions(task_id="unrelated", session_key="unrelated", include_retained=True) == []
|
||||
assert fresh.get("proc_0000") is None # Ambiguous across durable results.
|
||||
with scoped_current_session_id("unrelated-session"):
|
||||
assert load_completed_results(recovered.id) == {}
|
||||
assert fresh.get(recovered.id) is None
|
||||
from hermes_state import SessionDB
|
||||
db = SessionDB()
|
||||
try:
|
||||
db.create_session("owner-session", "cli")
|
||||
db.create_session("delegated-child", "subagent", parent_session_id="owner-session")
|
||||
with scoped_current_session_id("delegated-child"):
|
||||
assert fresh.get(recovered.id) is None
|
||||
db.end_session("owner-session", end_reason="compression")
|
||||
db.create_session("owner-tip", "cli", parent_session_id="owner-session")
|
||||
with scoped_current_session_id("owner-tip"):
|
||||
assert fresh.get(recovered.id).output_buffer == recovered.output_buffer
|
||||
finally:
|
||||
db.close()
|
||||
assert fresh.completion_queue.empty()
|
||||
for path in paths:
|
||||
expired = time.time() - receipts.RESULT_RETENTION_SECONDS - 1
|
||||
|
||||
@@ -739,9 +739,12 @@ class ProcessRegistry(ProcessCheckpointMixin):
|
||||
|
||||
@staticmethod
|
||||
def _new_session(command, task_id, owner_task_id, session_key, cwd, **extra) -> ProcessSession:
|
||||
from gateway.session_context import get_session_env
|
||||
|
||||
return ProcessSession(
|
||||
id=f"proc_{uuid.uuid4().hex[:12]}", command=command, task_id=task_id,
|
||||
owner_task_id=owner_task_id or task_id, session_key=session_key, cwd=cwd,
|
||||
parent_session_id=get_session_env("HERMES_SESSION_ID", ""),
|
||||
started_at=time.time(), **extra)
|
||||
|
||||
@staticmethod
|
||||
@@ -1926,7 +1929,7 @@ PROCESS_SCHEMA = {
|
||||
"description": (
|
||||
"Poll, wait on, or kill background terminal processes (from "
|
||||
"terminal(background=true)). "
|
||||
"Completed results remain retrievable by session_id after restart "
|
||||
"Completed results remain retrievable by session_id when resuming their owning conversation "
|
||||
"(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 "
|
||||
|
||||
@@ -8,6 +8,7 @@ parents cannot overwrite each other's results in the running-PID checkpoint.
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import sqlite3
|
||||
import time
|
||||
|
||||
from hermes_constants import get_hermes_home
|
||||
@@ -61,10 +62,29 @@ def save_completed_result(session) -> None:
|
||||
logger.warning("Could not retain completed process result %s", session.id, exc_info=True)
|
||||
|
||||
|
||||
def _owns_result(owner: str, parent: str | None) -> bool:
|
||||
if not parent:
|
||||
return False
|
||||
if owner == parent:
|
||||
return True
|
||||
from hermes_state import SessionDB
|
||||
|
||||
db = SessionDB()
|
||||
try:
|
||||
return db.get_compression_tip(parent) == owner
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def load_completed_results(prefix: str = "") -> dict:
|
||||
"""Restore read-only snapshots; no process handles, watchers, or queue events."""
|
||||
from tools.process_registry import ProcessSession
|
||||
|
||||
from gateway.session_context import get_session_env
|
||||
|
||||
owner = get_session_env("HERMES_SESSION_ID", "")
|
||||
if not owner:
|
||||
return {}
|
||||
results = {}
|
||||
try:
|
||||
paths = _result_paths()
|
||||
@@ -78,12 +98,14 @@ def load_completed_results(prefix: str = "") -> dict:
|
||||
record = json.loads(path.read_text(encoding="utf-8"))
|
||||
if record["id"] != path.stem or not re.fullmatch(r"proc_[\w]+", record["id"]):
|
||||
continue
|
||||
if not _owns_result(owner, record.get("parent_session_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):
|
||||
except (OSError, ValueError, KeyError, TypeError, sqlite3.Error):
|
||||
logger.debug("Skipping unreadable process result %s", path.name, exc_info=True)
|
||||
return results
|
||||
|
||||
@@ -231,7 +231,10 @@ writes one atomic, redacted receipt per process under the profile's
|
||||
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.
|
||||
notifications. Reads require the commissioning durable session or its compression
|
||||
continuation; knowing a handle alone does not authorize a retained result read.
|
||||
The registry captures that owner before starting any output reader, including on
|
||||
CLI and non-notifying processes. Receipt retention is bounded by age and count.
|
||||
|
||||
## Concurrency
|
||||
|
||||
|
||||
@@ -232,10 +232,12 @@ 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.
|
||||
active profile. Resume the conversation that launched the command (or its
|
||||
compressed continuation), then use the original `session_id` with
|
||||
`process(action="log")` for output and `process(action="poll")` for exit status.
|
||||
Unrelated conversations and requests without a bound owning session cannot read
|
||||
retained receipts, even with an exact process handle. `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
|
||||
|
||||
@@ -198,6 +198,21 @@ process(action="write", session_id="proc_abc123", data="y") # 发送输入
|
||||
|
||||
PTY 模式（`pty=true`）可启用 Codex 和 Claude Code 等交互式 CLI 工具。
|
||||
|
||||
## 已完成后台进程的结果
|
||||
|
||||
后台命令完成后，其退出状态和捕获的输出会保留在当前配置档案中。
|
||||
无头父进程退出或 Hermes 重启后，请恢复启动该命令的会话（或其上下文压缩后的
|
||||
延续会话），再使用原始 `session_id` 调用 `process(action="log")` 读取输出、
|
||||
调用 `process(action="poll")` 查看退出状态。即使知道完整进程标识，其他会话
|
||||
或未绑定所属会话的请求也不能读取保留的结果。`process(action="list")`
|
||||
也会列出当前任务或会话的保留结果。
|
||||
|
||||
每个配置档案在 `logs/process-results/` 下最多保留最近 **64 个已完成结果**，
|
||||
自完成起保存不超过 **7 天**。每份记录最多包含滚动输出末尾的 **200,000 个字符**，
|
||||
使用与终端输出相同的敏感信息脱敏规则。后续读取或写入结果时会清理过期记录。
|
||||
恢复结果不会重新运行命令，也不会重放完成通知。这仅保护父进程存活期间已经
|
||||
完成的工作，不保证未完成的子进程在超时或崩溃后继续运行。
|
||||
|
||||
## Sudo 支持
|
||||
|
||||
如果命令需要 sudo，系统会提示你输入密码（在本次会话内缓存）。也可在 `~/.hermes/.env` 中设置 `SUDO_PASSWORD`。
|
||||
|
||||
Reference in New Issue
Block a user