From 561a5ff3bcb69597e8092245c09b459cbbf79a0f Mon Sep 17 00:00:00 2001 From: CN-CLI Date: Mon, 20 Jul 2026 01:06:59 -0600 Subject: [PATCH 1/3] feat(agent): governed persistence funnel -- stage-or-deny via pre_persist_write New agent/persist_boundary.py::governed_persist(kind, path, content, meta) is the single chokepoint for runtime writes to Hermes' canonical memory/skill/ SOUL.md paths: it invokes the pre_persist_write REQUIRED hook (mirroring how cron/scheduler.py and hermes_cli/kanban_db.py invoke pre_run_start), and is the ONLY place that performs the actual disk write -- refactored call sites never touch a canonical path themselves. - allow+staged -> content quarantined governed-side; canonical path untouched. - block -> PersistResult(denied=True); call sites raise PersistDenied, the same clean-error contract a failed write already had. - bare allow (no persistence enforcer registered yet -- the pre-cutover default, mirroring the pre_run_start precedent) -> governed_persist performs the original atomic write itself, unchanged. - hook unreachable or an unrecognized directive -> decision-less local stage under $HERMES_HOME/persist-quarantine-local//, warned, grants nothing. hermes_cli/plugins.py: registers pre_persist_write as a valid required-hook boundary (VALID_HOOKS + register_required_hook's allowed set) and generalizes invoke_required_hook's context pass-through (previously run_id/session_id only) to also carry digest/staged, so a pre_persist_write directive's payload survives the hook aggregation. tools/memory_tool.py::MemoryStore._write_file and tools/skill_manager_tool.py::_atomic_write_text now route through governed_persist (kind="memory" / kind="skill" respectively); agent/learning_mutations.py shares the memory site's _write_file and is covered automatically. SOUL.md: recon found only bootstrap/default-creation writers and the dashboard's human-editor endpoint -- no runtime SOUL.md writer exists in this fork today; kind="soul" is implemented and tested for a future writer. Tests: tests/agent/test_persist_boundary.py covers the funnel's directive handling, the hook-unreachable local stage, both refactored write sites (deny/stage/passthrough/kind-mapping), and an AST-scoped source-scan pin asserting no canonical write primitive survives outside the funnel module. Co-Authored-By: Claude Fable 5 --- agent/persist_boundary.py | 290 +++++++++++++++++ hermes_cli/plugins.py | 34 +- tests/agent/test_persist_boundary.py | 471 +++++++++++++++++++++++++++ tools/memory_tool.py | 46 +-- tools/skill_manager_tool.py | 64 ++-- 5 files changed, 844 insertions(+), 61 deletions(-) create mode 100644 agent/persist_boundary.py create mode 100644 tests/agent/test_persist_boundary.py diff --git a/agent/persist_boundary.py b/agent/persist_boundary.py new file mode 100644 index 000000000000..093451b4b2b3 --- /dev/null +++ b/agent/persist_boundary.py @@ -0,0 +1,290 @@ +"""Governed persistence funnel -- the single chokepoint for runtime writes to +Hermes' canonical memory/skill/SOUL.md paths. + +Companion piece to agent-lineage's ``framework/adapters/hermes_persistence`` +(governed-runtime repo, external to this fork): every runtime mutation of a +memory, skill, or SOUL.md file is routed through :func:`governed_persist` +instead of writing its canonical path directly. This is the ONLY module in +this fork that performs the actual disk write for those paths -- refactored +call sites (``tools/memory_tool.py``, ``tools/skill_manager_tool.py``) hand +their content to ``governed_persist`` and never touch the canonical path +themselves; a source-scan test pins this (see +``tests/agent/test_persist_boundary.py``). + +``governed_persist`` invokes the ``pre_persist_write`` REQUIRED hook the same +way ``cron/scheduler.py`` and ``hermes_cli/kanban_db.py`` invoke +``pre_run_start`` -- via ``hermes_cli.plugins.get_required_hook_directive`` -- +and translates its directive into one of three outcomes: + +* **staged** (``action == "allow"`` with ``staged=True``): the content was + diverted into a governed, content-addressed quarantine by the + ``agent-lineage`` policy worker. The canonical path is NOT written here -- + release (verification + independent Ed25519 attestation + atomic promote) + is a separate, later governed step. Staging alone confers no authority. +* **denied** (``action == "block"``): the write is refused outright. + ``governed_persist`` returns ``PersistResult(denied=True, message=...)`` + without writing anything; each call site raises :class:`PersistDenied` + and lets it propagate exactly the way a failed write already did (the + tool registry's broad ``except Exception`` -> ``{"error": ...}`` dispatch + wrapper) -- no canonical write, no crash. +* **passthrough** (``action == "allow"`` with no ``staged`` flag): no + required-hook enforcer is currently registered for ``pre_persist_write`` + (the pre-cutover default -- see ``docs/HERMES_INTEGRATION.md`` and the + companion agent-lineage plan; deploying the governed install is an + explicit, separate cutover, not part of wiring this funnel). This mirrors + the established ``pre_run_start`` precedent (``cron/scheduler.py``, + ``hermes_cli/kanban_db.py``): a bare "allow" with nothing further attached + means "no governance is configured for this boundary yet" -- ``governed_ + persist`` performs the ORIGINAL atomic canonical write itself, unchanged. + Once an operator opts in (``plugins.required: [agent-lineage]``), the SAME + ``pre_persist_write`` callback starts returning ``staged=True`` and this + same code path enforces it with no further changes anywhere. + +When the hook itself is unreachable (the required-hook call raises -- for +example because the plugin subsystem isn't importable in this execution +context, or a worker crashes mid-call) OR returns something this funnel does +not recognize as a coherent decision, ``governed_persist`` makes NO policy +decision of its own. It stages the content locally, decision-less, under +``$HERMES_HOME/persist-quarantine-local//`` and logs a warning. +This grants nothing -- the canonical path is still never written on this +path -- and the audit gap is recorded (as a warning) for the next human or +operator to see; only a later governed sweep can ever release it. This is +deliberately safer than either raising a crash (losing the agent's turn) or +silently writing straight through (defeating the whole point of the funnel +during an outage). + +SOUL.md: recon across this fork (``hermes_cli/config.py``, +``hermes_cli/profiles.py``, ``hermes_cli/doctor.py``, and the dashboard's +``PUT /api/profiles/{name}/soul`` in ``hermes_cli/web_server.py``) found only +bootstrap/default-creation writers and the dashboard's human-editor endpoint +(a person editing SOUL.md through the desktop/web UI) -- no runtime +(agent-driven, autonomous) SOUL.md write site exists in this fork today. Per +the design's scope decision #1, human editor edits to SOUL.md are explicitly +OUT of scope for this funnel by construction (there is no runtime hook to +traverse). ``governed_persist("soul", ...)`` is fully implemented and tested +so a future runtime SOUL writer only has to call it -- no funnel changes +needed -- but nothing calls it with ``kind="soul"`` yet. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import logging +import os +import tempfile +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from hermes_constants import get_hermes_home +from utils import atomic_replace + +__all__ = ["PersistResult", "PersistDenied", "governed_persist"] + +logger = logging.getLogger(__name__) + +VALID_KINDS = frozenset({"memory", "skill", "soul"}) + +_LOCAL_STAGE_DIRNAME = "persist-quarantine-local" +_DEFAULT_SESSION_ID = "hermes.persist.local" + + +class PersistDenied(RuntimeError): + """Raised by a refactored write site when ``governed_persist`` denies a write. + + ``governed_persist`` itself only RETURNS ``PersistResult(denied=True, ...)`` + -- it never raises. Each call site is responsible for turning a denial + into this exception and letting it propagate to whatever already turns + an exception from the old direct write into a caller-visible error (the + tool registry's broad ``except Exception`` -> ``{"error": ...}`` dispatch + wrapper) -- exactly mirroring how an ``OSError`` from the previous direct + write used to surface. + """ + + +@dataclass(frozen=True) +class PersistResult: + """Outcome of routing one write through the governed persistence funnel. + + The canonical write, when one happens at all, is already DONE by the + time this is returned -- ``governed_persist`` is the only place that + performs it (directly on passthrough, never on ``staged``/``denied``). + """ + + staged: bool + digest: Optional[str] + denied: bool + message: str = "" + + +def _atomic_write_bytes(path: Path, content: bytes, *, prefix: str) -> None: + """Temp-file + atomic rename, shared by the canonical write and the + decision-less local stage. fsync's before rename for durability.""" + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), prefix=prefix, suffix=".tmp") + try: + with os.fdopen(fd, "wb") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + atomic_replace(tmp_path, path) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + pass + raise + + +def _hook_target_path(file_path: Path) -> str: + """Best-effort path, relative to HERMES_HOME, to send as the hook's + ``target_path`` -- purely descriptive/audit metadata for the eventual + governed release step. Falls back to the bare filename when the target + lives outside HERMES_HOME (e.g. an externally-configured skills root); + the actual write below always uses the caller's exact ``file_path``, so + this fallback never affects where content is written. + """ + try: + home = get_hermes_home().resolve() + return file_path.resolve().relative_to(home).as_posix() + except (OSError, ValueError): + return file_path.name + + +def _stage_local(kind: str, target_path: str, content: bytes, meta: dict[str, Any]) -> PersistResult: + """Decision-less local fallback when the required hook is unreachable. + + Grants nothing: the content is durably recorded so it isn't lost, but + only a later governed sweep (release, agent-lineage side) can ever + promote it to a canonical path. + """ + hex_digest = hashlib.sha256(content).hexdigest() + root = get_hermes_home() / _LOCAL_STAGE_DIRNAME / hex_digest + record = { + "kind": kind, + "target_path": target_path, + "digest": f"sha256:{hex_digest}", + "meta": meta, + "staged_at": datetime.now(timezone.utc).isoformat(), + "status": "staged-local", + } + try: + _atomic_write_bytes(root / "content.bin", content, prefix=".persist-local-content-") + _atomic_write_bytes( + root / "meta.json", + (json.dumps(record, indent=2, sort_keys=True) + "\n").encode("utf-8"), + prefix=".persist-local-meta-", + ) + except OSError as exc: + logger.error( + "governed_persist: local staging failed for kind=%s (%s) -- refusing the write", + kind, exc, + ) + return PersistResult( + staged=False, + digest=None, + denied=True, + message="persistence unavailable: local staging failed", + ) + logger.warning( + "governed_persist: pre_persist_write is unreachable -- staged '%s' (kind=%s) " + "locally under %s with no policy decision; canonical path was NOT written", + target_path, kind, root, + ) + return PersistResult( + staged=True, + digest=f"sha256:{hex_digest}", + denied=False, + message="staged locally: no policy enforcer reachable for pre_persist_write", + ) + + +def governed_persist( + kind: str, + path: str, + content: "bytes | str", + meta: Optional[dict[str, Any]] = None, +) -> PersistResult: + """Route one runtime write through the ``pre_persist_write`` funnel. + + ``path`` is the actual canonical filesystem path the runtime would have + written to (absolute, or resolvable from the process's cwd). Returns a + :class:`PersistResult` describing what happened; raises + :class:`PersistDenied` when the write was refused. The canonical write + itself -- when one happens at all -- is performed HERE, atomically, and + only on the passthrough outcome; ``staged``/``denied`` never touch it. + """ + if kind not in VALID_KINDS: + raise ValueError(f"kind must be one of {sorted(VALID_KINDS)}, got {kind!r}") + if not isinstance(path, str) or not path.strip(): + raise ValueError("path must be a non-empty string") + if isinstance(content, str): + content_bytes = content.encode("utf-8") + elif isinstance(content, (bytes, bytearray)): + content_bytes = bytes(content) + else: + raise TypeError("content must be bytes or str") + + file_path = Path(path) + target_path = _hook_target_path(file_path) + meta_dict: dict[str, Any] = dict(meta) if isinstance(meta, dict) else {} + session_id = str(meta_dict.get("session_id") or _DEFAULT_SESSION_ID)[:256] + + try: + from hermes_cli.plugins import get_required_hook_directive + + directive = get_required_hook_directive( + "pre_persist_write", + session_id=session_id, + kind=kind, + target_path=target_path, + content_b64=base64.b64encode(content_bytes).decode("ascii"), + meta=meta_dict, + ) + except Exception as exc: + logger.warning( + "governed_persist: pre_persist_write hook unreachable (%s: %s) -- " + "staging kind=%s locally with no governance decision", + type(exc).__name__, exc, kind, + ) + return _stage_local(kind, target_path, content_bytes, meta_dict) + + if not isinstance(directive, dict): + logger.warning( + "governed_persist: pre_persist_write returned a non-dict directive " + "(%r) -- staging kind=%s locally with no governance decision", + directive, kind, + ) + return _stage_local(kind, target_path, content_bytes, meta_dict) + + action = directive.get("action") + + if action == "block": + message = str(directive.get("message") or "persistence denied by policy") + return PersistResult(staged=False, digest=None, denied=True, message=message) + + if action == "allow" and directive.get("staged") is True: + digest = directive.get("digest") + digest = digest if isinstance(digest, str) and digest else None + return PersistResult(staged=True, digest=digest, denied=False, message="") + + if action == "allow": + # No required enforcer is registered for pre_persist_write (the + # pre-cutover default) -- perform the original canonical write + # ourselves. See the module docstring's "passthrough" case. + _atomic_write_bytes(file_path, content_bytes, prefix=f".{file_path.name}.persist.") + return PersistResult(staged=False, digest=None, denied=False, message="") + + # Anything else -- a missing action, or a directive shape this funnel + # does not recognize as a coherent decision ("approve" is not valid for + # pre_persist_write) -- is not something to guess about. Fail closed to + # the same decision-less local stage as an unreachable hook. + logger.warning( + "governed_persist: pre_persist_write returned an unrecognized directive " + "%r -- staging kind=%s locally with no governance decision", + directive, kind, + ) + return _stage_local(kind, target_path, content_bytes, meta_dict) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index f4f0ed1f0521..5a000e1b15e9 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -166,6 +166,14 @@ def _install_plugin_debug_handler(force: bool = False) -> None: # plus an observer emitted after an admitted run finishes. "pre_run_start", "post_run_end", + # Mandatory admission before a runtime mutation of a memory/skill/SOUL.md + # file reaches its canonical path. The chokepoint (agent.persist_boundary + # .governed_persist) sends {session_id, kind, target_path, content_b64, + # meta}; the callback returns an "allow" directive that MAY carry + # {"digest": ..., "staged": True} to mean the content was diverted into a + # governed quarantine (canonical path left untouched -- release is a + # separate, later governed step), or "block" to refuse the write outright. + "pre_persist_write", "on_session_start", "on_session_end", "on_session_finalize", @@ -1239,9 +1247,10 @@ def register_required_hook(self, hook_name: str, callback: Callable) -> None: Required hooks are limited to boundaries where Hermes can prevent spend or stop the response from advancing: tool calls, model request - reservation/settlement, and scheduler-owned run admission. The - callback must return an explicit ``allow``, ``approve``, or ``block`` - directive. Exceptions and malformed/empty returns fail closed. + reservation/settlement, scheduler-owned run admission, and runtime + persistence writes. The callback must return an explicit ``allow``, + ``approve``, or ``block`` directive. Exceptions and malformed/empty + returns fail closed. Operators make the plugin itself mandatory with ``plugins.required: []``. A required plugin that is missing, @@ -1253,6 +1262,7 @@ def register_required_hook(self, hook_name: str, callback: Callable) -> None: "pre_api_request", "post_api_request", "pre_run_start", + "pre_persist_write", }: raise ValueError( "required hook is not an enforceable Hermes boundary" @@ -2145,7 +2155,7 @@ def invoke_required_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: return [] approval: dict[str, Any] | None = None - context: dict[str, str] = {} + context: dict[str, Any] = {} for plugin_id, callback in active_callbacks: try: result = callback(**kwargs) @@ -2174,9 +2184,21 @@ def invoke_required_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: "message": message if isinstance(message, str) and message else "BLOCKED: required policy enforcer denied this operation", }] - for key in ("run_id", "session_id"): + # run_id/session_id carry a scheduler-owned run binding through + # (pre_run_start); digest/staged carry a quarantine binding + # through (pre_persist_write) -- same pass-through contract, + # generalized so a caller-facing directive from get_required_hook_ + # directive() isn't silently stripped down to the bare action. + for key, expected_type in ( + ("run_id", str), + ("session_id", str), + ("digest", str), + ("staged", bool), + ): value = result.get(key) - if not isinstance(value, str) or not value: + if not isinstance(value, expected_type): + continue + if expected_type is str and not value: continue prior = context.get(key) if prior is not None and prior != value: diff --git a/tests/agent/test_persist_boundary.py b/tests/agent/test_persist_boundary.py new file mode 100644 index 000000000000..3068c30d6da7 --- /dev/null +++ b/tests/agent/test_persist_boundary.py @@ -0,0 +1,471 @@ +"""Tests for agent.persist_boundary -- the governed persistence funnel. + +Covers the funnel's own decision logic (``governed_persist``), its wiring +into the two real write sites it replaced (``tools/memory_tool.py:: +MemoryStore._write_file``, ``tools/skill_manager_tool.py::_atomic_write_text`` +-- ``agent/learning_mutations.py`` shares the memory site's underlying +``_write_file``, so it is covered by the same funnel automatically), and an +AST-scoped source-scan pin asserting no canonical memory/skill write +primitive survives outside the funnel module. + +Mirrors the monkeypatched-``get_required_hook_directive`` style of +tests/hermes_cli/test_required_enforcers.py: the funnel calls +``hermes_cli.plugins.get_required_hook_directive("pre_persist_write", ...)`` +exactly the way cron/scheduler.py and hermes_cli/kanban_db.py call it for +``pre_run_start`` -- no real plugin runtime is loaded here, the directive is +supplied directly. +""" + +from __future__ import annotations + +import ast +import hashlib +import json +from pathlib import Path + +import pytest + +from agent import persist_boundary as pb +from agent.persist_boundary import PersistDenied, PersistResult, governed_persist + + +def _directive(action, **extra): + def _fake(hook_name, **kwargs): + assert hook_name == "pre_persist_write" + return {"action": action, **extra} + return _fake + + +# --------------------------------------------------------------------------- +# governed_persist -- input validation +# --------------------------------------------------------------------------- + + +class TestGovernedPersistValidation: + def test_bad_kind_raises(self, tmp_path): + with pytest.raises(ValueError): + governed_persist("bogus", str(tmp_path / "x.md"), b"content") + + def test_empty_path_raises(self): + with pytest.raises(ValueError): + governed_persist("memory", "", b"content") + + def test_non_string_path_raises(self): + with pytest.raises(ValueError): + governed_persist("memory", None, b"content") # type: ignore[arg-type] + + def test_bad_content_type_raises(self, tmp_path): + with pytest.raises(TypeError): + governed_persist("memory", str(tmp_path / "x.md"), 12345) # type: ignore[arg-type] + + def test_str_content_is_utf8_encoded(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + seen = {} + + def fake(hook_name, **kwargs): + seen["content_b64"] = kwargs["content_b64"] + return {"action": "allow"} + + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", fake) + governed_persist("memory", str(target), "héllo") + import base64 + assert base64.b64decode(seen["content_b64"]) == "héllo".encode("utf-8") + + +# --------------------------------------------------------------------------- +# governed_persist -- directive handling +# --------------------------------------------------------------------------- + + +class TestGovernedPersistDirectives: + def test_deny_blocks_persistence_no_canonical_write(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("block", message="POLICY_DENIED: refused"), + ) + + result = governed_persist("memory", str(target), "some content") + + assert result == PersistResult(staged=False, digest=None, denied=True, message="POLICY_DENIED: refused") + assert not target.exists() + + def test_deny_without_message_gets_a_default(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("block")) + + result = governed_persist("memory", str(target), b"x") + + assert result.denied is True + assert result.message + assert not target.exists() + + def test_allow_staged_leaves_canonical_absent(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + digest = "sha256:" + "a" * 64 + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("allow", staged=True, digest=digest), + ) + + result = governed_persist("memory", str(target), b"quarantined content") + + assert result == PersistResult(staged=True, digest=digest, denied=False, message="") + assert not target.exists() + + def test_bare_allow_is_passthrough_and_writes_canonical(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("allow")) + + result = governed_persist("memory", str(target), "hello") + + assert result == PersistResult(staged=False, digest=None, denied=False, message="") + assert target.read_text(encoding="utf-8") == "hello" + + def test_passthrough_writes_bytes_content_exactly(self, tmp_path, monkeypatch): + target = tmp_path / "SKILL.md" + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("allow")) + + governed_persist("skill", str(target), b"\x00binary-ish\xff") + + assert target.read_bytes() == b"\x00binary-ish\xff" + + def test_passthrough_creates_parent_directories(self, tmp_path, monkeypatch): + target = tmp_path / "nested" / "dir" / "MEMORY.md" + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("allow")) + + governed_persist("memory", str(target), "x") + + assert target.exists() + + +# --------------------------------------------------------------------------- +# governed_persist -- hook unreachable / unrecognized -> decision-less local stage +# --------------------------------------------------------------------------- + + +class TestGovernedPersistHookUnreachable: + def test_hook_raises_stages_locally_with_no_canonical_write(self, tmp_path, monkeypatch, caplog): + target = tmp_path / "MEMORY.md" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + def boom(hook_name, **kwargs): + raise RuntimeError("plugin subsystem exploded") + + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", boom) + + with caplog.at_level("WARNING"): + result = governed_persist("memory", str(target), b"unsaved content", meta={"session_id": "s1"}) + + assert result.staged is True + assert result.denied is False + assert result.digest == "sha256:" + hashlib.sha256(b"unsaved content").hexdigest() + assert not target.exists() + assert "pre_persist_write hook unreachable" in caplog.text + + staged_dir = tmp_path / "persist-quarantine-local" / hashlib.sha256(b"unsaved content").hexdigest() + assert staged_dir.joinpath("content.bin").read_bytes() == b"unsaved content" + meta = json.loads(staged_dir.joinpath("meta.json").read_text(encoding="utf-8")) + assert meta["kind"] == "memory" + assert meta["status"] == "staged-local" + assert meta["meta"]["session_id"] == "s1" + + def test_unrecognized_directive_stages_locally(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("approve", message="needs a human"), + ) + + result = governed_persist("memory", str(target), b"content") + + assert result.staged is True + assert result.denied is False + assert not target.exists() + + def test_non_dict_directive_stages_locally(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", lambda hook_name, **kw: None) + + result = governed_persist("memory", str(target), b"content") + + assert result.staged is True + assert not target.exists() + + def test_local_stage_is_idempotent_by_digest(self, tmp_path, monkeypatch): + target = tmp_path / "MEMORY.md" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + lambda hook_name, **kw: (_ for _ in ()).throw(RuntimeError("down")), + ) + + first = governed_persist("memory", str(target), b"same content") + second = governed_persist("memory", str(target), b"same content") + + assert first.digest == second.digest + + +# --------------------------------------------------------------------------- +# Refactored site: tools/memory_tool.py (kind="memory") +# --------------------------------------------------------------------------- + + +class TestMemoryToolSiteWiring: + def test_add_denied_raises_persist_denied_no_canonical_write(self, tmp_path, monkeypatch): + from tools.memory_tool import MemoryStore + + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("block", message="POLICY_DENIED: nope"), + ) + + store = MemoryStore() + store.load_from_disk() + with pytest.raises(PersistDenied): + store.add("memory", "should never land on disk") + + assert not (tmp_path / "MEMORY.md").exists() + + def test_add_staged_leaves_canonical_absent(self, tmp_path, monkeypatch): + from tools.memory_tool import MemoryStore + + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("allow", staged=True, digest="sha256:" + "c" * 64), + ) + + store = MemoryStore() + store.load_from_disk() + result = store.add("memory", "quarantined text") + + assert result["success"] is True + assert not (tmp_path / "MEMORY.md").exists() + + def test_add_passthrough_still_writes_canonical(self, tmp_path, monkeypatch): + from tools.memory_tool import MemoryStore + + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("allow")) + + store = MemoryStore() + store.load_from_disk() + store.add("memory", "regular entry") + + assert "regular entry" in (tmp_path / "MEMORY.md").read_text(encoding="utf-8") + + def test_add_routes_kind_memory(self, tmp_path, monkeypatch): + from tools.memory_tool import MemoryStore + + monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path) + seen = {} + + def fake_governed_persist(kind, path, content, meta=None): + seen["kind"] = kind + seen["path"] = path + return PersistResult(staged=False, digest=None, denied=False) + + monkeypatch.setattr("agent.persist_boundary.governed_persist", fake_governed_persist) + + store = MemoryStore() + store.load_from_disk() + store.add("memory", "hello") + + assert seen["kind"] == "memory" + assert seen["path"].endswith("MEMORY.md") + + +# --------------------------------------------------------------------------- +# Refactored site: agent/learning_mutations.py (shares MemoryStore._write_file) +# --------------------------------------------------------------------------- + + +class TestLearningMutationsSiteWiring: + def test_edit_memory_denied_raises_persist_denied(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / "memories").mkdir(parents=True, exist_ok=True) + (tmp_path / "memories" / "MEMORY.md").write_text("alpha\n§\nbeta", encoding="utf-8") + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("block", message="POLICY_DENIED: nope"), + ) + + from agent import learning_mutations as lm + + with pytest.raises(PersistDenied): + lm.edit_node("memory:memory:0", "replacement text") + + assert "alpha" in (tmp_path / "memories" / "MEMORY.md").read_text(encoding="utf-8") + + +# --------------------------------------------------------------------------- +# Refactored site: tools/skill_manager_tool.py (kind="skill") +# --------------------------------------------------------------------------- + +_VALID_SKILL = """\ +--- +name: my-skill +description: A test skill for the persist boundary. +--- + +# My Skill + +Step 1: Do the thing. +""" + + +class TestSkillManagerSiteWiring: + def _patched(self, tmp_path, monkeypatch): + monkeypatch.setattr("tools.skill_manager_tool.SKILLS_DIR", tmp_path) + monkeypatch.setattr("agent.skill_utils.get_all_skills_dirs", lambda: [tmp_path]) + + def test_create_denied_raises_persist_denied_no_skill_md(self, tmp_path, monkeypatch): + self._patched(tmp_path, monkeypatch) + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("block", message="POLICY_DENIED: nope"), + ) + + from tools.skill_manager_tool import _create_skill + + with pytest.raises(PersistDenied): + _create_skill("my-skill", _VALID_SKILL) + + assert not (tmp_path / "my-skill" / "SKILL.md").exists() + + def test_create_staged_leaves_skill_md_absent(self, tmp_path, monkeypatch): + self._patched(tmp_path, monkeypatch) + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("allow", staged=True, digest="sha256:" + "d" * 64), + ) + + from tools.skill_manager_tool import _create_skill + + # governed_persist doesn't raise on staged, so _create_skill runs its + # post-write steps (security scan) against a file that was never + # written -- documented residual of wiring the funnel ahead of the + # governed install cutover (see agent/persist_boundary.py docstring). + # The one invariant under test is the one that matters here: no + # unattested content ever reaches the canonical path. + try: + _create_skill("my-skill", _VALID_SKILL) + except Exception: + pass + + assert not (tmp_path / "my-skill" / "SKILL.md").exists() + + def test_create_passthrough_still_writes_canonical(self, tmp_path, monkeypatch): + self._patched(tmp_path, monkeypatch) + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("allow")) + + from tools.skill_manager_tool import _create_skill + + result = _create_skill("my-skill", _VALID_SKILL) + + assert result["success"] is True + assert (tmp_path / "my-skill" / "SKILL.md").read_text(encoding="utf-8") == _VALID_SKILL + + def test_create_routes_kind_skill(self, tmp_path, monkeypatch): + self._patched(tmp_path, monkeypatch) + seen = {} + + def fake_governed_persist(kind, path, content, meta=None): + seen["kind"] = kind + seen["path"] = path + return PersistResult(staged=False, digest=None, denied=False) + + monkeypatch.setattr("agent.persist_boundary.governed_persist", fake_governed_persist) + + from tools.skill_manager_tool import _create_skill + + _create_skill("my-skill", _VALID_SKILL) + + assert seen["kind"] == "skill" + assert seen["path"].endswith("SKILL.md") + + +# --------------------------------------------------------------------------- +# Grep-pin: only the funnel module may perform a canonical write +# +# Scoped to the specific write-site FUNCTIONS (via AST), not a whole-file +# text scan -- tools/memory_tool.py legitimately writes a `.bak.` DRIFT +# snapshot elsewhere in the same file (external-drift recovery, an +# operator-facing diagnostic copy, not a canonical read path Hermes ever +# loads from) and that write is intentionally NOT part of this funnel +# ("DO NOT refactor unrelated writes"). Pinning the exact write-site +# function bodies is the precise version of "no direct canonical write +# outside the funnel module." +# --------------------------------------------------------------------------- + +_FORBIDDEN_WRITE_SNIPPETS = ( + "os.fdopen(", + "tempfile.mkstemp(", + "atomic_replace(", + ".write_text(", +) + +_WRITE_SITE_FUNCTIONS = ( + ("tools/memory_tool.py", "_write_file"), + ("tools/skill_manager_tool.py", "_atomic_write_text"), +) + +_WRITE_SITE_WHOLE_FILES = ( + # No write primitive of its own -- delegates entirely to + # MemoryStore._write_file (pinned above). Whole-file scan is safe here: + # unlike memory_tool.py, this module has no unrelated diagnostic write. + "agent/learning_mutations.py", +) + +_FUNNEL_MODULE = "agent/persist_boundary.py" + + +def _function_source(repo_root: Path, rel_path: str, func_name: str) -> str: + file_path = repo_root / rel_path + source = file_path.read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(file_path)) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == func_name: + segment = ast.get_source_segment(source, node) + if segment is not None: + return segment + raise AssertionError(f"{func_name!r} not found in {rel_path}") + + +class TestNoDirectCanonicalWriteOutsideTheFunnel: + def _repo_root(self) -> Path: + return Path(__file__).resolve().parents[2] + + def test_write_site_functions_contain_no_direct_write_primitive(self): + repo_root = self._repo_root() + for rel, func_name in _WRITE_SITE_FUNCTIONS: + segment = _function_source(repo_root, rel, func_name) + for snippet in _FORBIDDEN_WRITE_SNIPPETS: + assert snippet not in segment, ( + f"{rel}::{func_name} still contains a direct canonical " + f"write primitive ({snippet!r}) -- route it through " + f"agent.persist_boundary.governed_persist instead" + ) + + def test_delegating_modules_contain_no_direct_write_primitive(self): + repo_root = self._repo_root() + for rel in _WRITE_SITE_WHOLE_FILES: + source = (repo_root / rel).read_text(encoding="utf-8") + for snippet in _FORBIDDEN_WRITE_SNIPPETS: + assert snippet not in source, ( + f"{rel} contains a direct canonical write primitive " + f"({snippet!r}) -- route it through agent.persist_boundary." + f"governed_persist instead" + ) + + def test_funnel_module_is_the_one_allowed_writer(self): + # Sanity check the allowlist isn't hiding a funnel that silently lost + # its own ability to ever write a canonical path. + repo_root = self._repo_root() + source = (repo_root / _FUNNEL_MODULE).read_text(encoding="utf-8") + assert "os.fdopen(" in source + assert "atomic_replace(" in source diff --git a/tools/memory_tool.py b/tools/memory_tool.py index 08eeaa470ea4..dbb008c48735 100644 --- a/tools/memory_tool.py +++ b/tools/memory_tool.py @@ -25,16 +25,12 @@ import json import logging -import os -import tempfile import time from contextlib import contextmanager from pathlib import Path from hermes_constants import get_hermes_home from typing import Dict, Any, List, Optional -from utils import atomic_replace - # fcntl is Unix-only; on Windows use msvcrt for file locking msvcrt = None try: @@ -758,34 +754,24 @@ def _detect_external_drift(self, target: str) -> Optional[str]: @staticmethod def _write_file(path: Path, entries: List[str]): - """Write entries to a memory file using atomic temp-file + rename. - - Previous implementation used open("w") + flock, but "w" truncates the - file *before* the lock is acquired, creating a race window where - concurrent readers see an empty file. Atomic rename avoids this: - readers always see either the old complete file or the new one. + """Persist entries to a memory file through the governed persistence + funnel (agent.persist_boundary.governed_persist, kind="memory"). + + This is now the ONLY place a memory file's canonical path can be + written from this fork -- ``governed_persist`` performs the actual + atomic temp-file + rename write itself (on the pre-cutover + passthrough outcome); a denial raises :class:`PersistDenied` (the + same clean-error contract a previous write failure already had for + callers), and a staged outcome leaves the canonical path untouched + (release is a separate, later governed step -- see + docs/HERMES_INTEGRATION.md). """ + from agent.persist_boundary import PersistDenied, governed_persist + content = ENTRY_DELIMITER.join(entries) if entries else "" - try: - # Write to temp file in same directory (same filesystem for atomic rename) - fd, tmp_path = tempfile.mkstemp( - dir=str(path.parent), suffix=".tmp", prefix=".mem_" - ) - try: - with os.fdopen(fd, "w", encoding="utf-8") as f: - f.write(content) - f.flush() - os.fsync(f.fileno()) - atomic_replace(tmp_path, path) - except BaseException: - # Clean up temp file on any failure - try: - os.unlink(tmp_path) - except OSError: - pass - raise - except (OSError, IOError) as e: - raise RuntimeError(f"Failed to write memory file {path}: {e}") + result = governed_persist("memory", str(path), content, meta={}) + if result.denied: + raise PersistDenied(result.message) def load_on_disk_store() -> "MemoryStore": diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py index debea52642f9..d36915f8059d 100644 --- a/tools/skill_manager_tool.py +++ b/tools/skill_manager_tool.py @@ -34,16 +34,14 @@ import json import logging -import os import re import shutil -import tempfile import contextvars as _ctxvars from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from hermes_constants import get_hermes_home, display_hermes_home -from utils import atomic_replace, is_truthy_value +from utils import is_truthy_value from hermes_cli.config import cfg_get logger = logging.getLogger(__name__) @@ -774,36 +772,52 @@ def _resolve_skill_target(skill_dir: Path, file_path: str) -> Tuple[Optional[Pat return target, None +def _skill_target_path(file_path: Path) -> str: + """Return ``file_path`` relative to its containing skills root, POSIX-separated. + + Purely descriptive metadata forwarded to the governed persistence funnel + (the eventual release-time promotion step uses it to know where the + content belongs). A path that doesn't resolve under any known skills + root falls back to its last two components -- still a useful audit hint. + """ + root = _containing_skills_root(file_path) + try: + return file_path.resolve().relative_to(root.resolve()).as_posix() + except (OSError, ValueError): + parts = file_path.parts[-2:] + return "/".join(parts) if len(parts) == 2 else file_path.name + + def _atomic_write_text(file_path: Path, content: str, encoding: str = "utf-8") -> None: """ - Atomically write text content to a file. - - Uses a temporary file in the same directory and os.replace() to ensure - the target file is never left in a partially-written state if the process - crashes or is interrupted. - + Write text content to a skill file through the governed persistence + funnel (agent.persist_boundary.governed_persist, kind="skill"). + + This is now the ONLY place a skill file's canonical path can be written + from this fork -- covers create/edit/patch/write_file and their + rollback-to-original writes after a failed security scan uniformly + ("canonical paths are never written by the runtime" applies even to a + revert). ``governed_persist`` performs the actual atomic temp-file + + rename write itself (on the pre-cutover passthrough outcome); a denial + raises :class:`PersistDenied` (the same clean-error contract a previous + write failure already had for callers -- propagates to the tool + registry's broad exception -> error-JSON dispatch wrapper), and a staged + outcome leaves the canonical path untouched (release is a separate, + later governed step -- see docs/HERMES_INTEGRATION.md). + Args: file_path: Target file path content: Content to write encoding: Text encoding (default: utf-8) """ - file_path.parent.mkdir(parents=True, exist_ok=True) - fd, temp_path = tempfile.mkstemp( - dir=str(file_path.parent), - prefix=f".{file_path.name}.tmp.", - suffix="", + from agent.persist_boundary import PersistDenied, governed_persist + + result = governed_persist( + "skill", str(file_path), content.encode(encoding), + meta={"encoding": encoding, "skill_relative_path": _skill_target_path(file_path)}, ) - try: - with os.fdopen(fd, "w", encoding=encoding) as f: - f.write(content) - atomic_replace(temp_path, file_path) - except Exception: - # Clean up temp file on error - try: - os.unlink(temp_path) - except OSError: - logger.error("Failed to remove temporary file %s during atomic write", temp_path, exc_info=True) - raise + if result.denied: + raise PersistDenied(result.message) # ============================================================================= From 955f82e1dcad70b5c202d83330dc2d73da157748 Mon Sep 17 00:00:00 2001 From: CN-CLI Date: Mon, 20 Jul 2026 06:32:56 -0600 Subject: [PATCH 2/3] fix(agent): fail-closed on malformed enforcer allow + hardened write-site grep pin --- agent/persist_boundary.py | 65 ++++++++++++- tests/agent/test_persist_boundary.py | 137 +++++++++++++++++++++++++++ 2 files changed, 197 insertions(+), 5 deletions(-) diff --git a/agent/persist_boundary.py b/agent/persist_boundary.py index 093451b4b2b3..7b856bbde3ca 100644 --- a/agent/persist_boundary.py +++ b/agent/persist_boundary.py @@ -27,11 +27,11 @@ and lets it propagate exactly the way a failed write already did (the tool registry's broad ``except Exception`` -> ``{"error": ...}`` dispatch wrapper) -- no canonical write, no crash. -* **passthrough** (``action == "allow"`` with no ``staged`` flag): no - required-hook enforcer is currently registered for ``pre_persist_write`` - (the pre-cutover default -- see ``docs/HERMES_INTEGRATION.md`` and the - companion agent-lineage plan; deploying the governed install is an - explicit, separate cutover, not part of wiring this funnel). This mirrors +* **passthrough** (``action == "allow"`` with no ``staged`` flag, AND no + required-hook enforcer is actually registered for ``pre_persist_write``): + this is the pre-cutover default -- see ``docs/HERMES_INTEGRATION.md`` and + the companion agent-lineage plan; deploying the governed install is an + explicit, separate cutover, not part of wiring this funnel. This mirrors the established ``pre_run_start`` precedent (``cron/scheduler.py``, ``hermes_cli/kanban_db.py``): a bare "allow" with nothing further attached means "no governance is configured for this boundary yet" -- ``governed_ @@ -40,6 +40,18 @@ ``pre_persist_write`` callback starts returning ``staged=True`` and this same code path enforces it with no further changes anywhere. + Critically, "no ``staged`` flag" alone is NOT enough to earn passthrough: + ``governed_persist`` also checks (via ``_enforcer_registered``) whether a + required-hook callback is actually wired for ``pre_persist_write`` right + now. If one IS registered and still returns a bare "allow" -- a malformed + or buggy enforcer, since ``invoke_required_hook``'s own type guard silently + drops a non-``bool`` or missing ``staged`` key rather than blocking -- that + is indistinguishable, by directive shape alone, from "no enforcer at all". + Treating it as passthrough would let a broken enforcer silently disable + itself. So when an enforcer IS present, a bare allow is instead routed to + the same decision-less local stage as an unreachable/unrecognized + directive (see below) -- fail-closed, never a silent canonical write. + When the hook itself is unreachable (the required-hook call raises -- for example because the plugin subsystem isn't importable in this execution context, or a worker crashes mid-call) OR returns something this funnel does @@ -202,6 +214,35 @@ def _stage_local(kind: str, target_path: str, content: bytes, meta: dict[str, An ) +def _enforcer_registered(hook_name: str) -> bool: + """True when at least one plugin has called ``register_required_hook`` + for *hook_name* right now. + + Mirrors how ``PluginManager._validate_required_plugins`` (hermes_cli/ + plugins.py) answers the same question for ``pre_tool_call`` -- it reads + the manager's ``_required_hooks`` registry directly. ``plugins.has_hook`` + cannot be used here: it only sees the OPTIONAL hook registry populated by + ``register_hook``, never the required-hook registry populated by + ``register_required_hook`` -- a real pre_persist_write enforcer would + never show up in it. There is no dedicated public accessor for the + required-hook registry today, so this reaches into the same manager + internals the plugin runtime's own validation code does. + + Best-effort: any failure introspecting plugin-runtime internals is + treated as "an enforcer IS present" -- the safer default for what this + predicate gates (a malformed/unrecognized allow staging locally instead + of writing canonical), matching this module's fail-closed posture + everywhere else. + """ + try: + from hermes_cli.plugins import get_plugin_manager + + manager = get_plugin_manager() + return bool(manager._required_hooks.get(hook_name)) + except Exception: + return True + + def governed_persist( kind: str, path: str, @@ -272,6 +313,20 @@ def governed_persist( return PersistResult(staged=True, digest=digest, denied=False, message="") if action == "allow": + if _enforcer_registered("pre_persist_write"): + # A required enforcer IS wired for this boundary but returned a + # bare "allow" (missing/non-bool `staged` -- invoke_required_ + # hook's type guard strips it rather than blocking). That is + # NOT the pre-cutover passthrough case; it's a malformed + # enforcer directive. Never guess it into a canonical write -- + # fail closed to the same decision-less local stage as an + # unreachable/unrecognized hook. + logger.warning( + "governed_persist: pre_persist_write enforcer allow without " + "staged -- staging locally, fail-closed (kind=%s, target=%s)", + kind, target_path, + ) + return _stage_local(kind, target_path, content_bytes, meta_dict) # No required enforcer is registered for pre_persist_write (the # pre-cutover default) -- perform the original canonical write # ourselves. See the module docstring's "passthrough" case. diff --git a/tests/agent/test_persist_boundary.py b/tests/agent/test_persist_boundary.py index 3068c30d6da7..031382724d70 100644 --- a/tests/agent/test_persist_boundary.py +++ b/tests/agent/test_persist_boundary.py @@ -139,6 +139,83 @@ def test_passthrough_creates_parent_directories(self, tmp_path, monkeypatch): assert target.exists() +# --------------------------------------------------------------------------- +# governed_persist -- enforcer presence gates a bare "allow" +# +# A bare "allow" (no `staged` flag) is only a legitimate pre-cutover +# passthrough when NO required-hook enforcer is wired for pre_persist_write. +# When one IS wired, invoke_required_hook's own type guard silently strips a +# missing/non-bool `staged` key rather than blocking, so the resulting +# directive is shape-identical to the no-enforcer case -- governed_persist +# must tell the two apart itself (via _enforcer_registered) and fail closed +# to a local stage rather than guessing it into a canonical write. +# --------------------------------------------------------------------------- + + +class TestGovernedPersistEnforcerPresence: + def test_enforcer_present_bare_allow_stages_locally_not_canonical( + self, tmp_path, monkeypatch, caplog + ): + target = tmp_path / "MEMORY.md" + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("allow")) + monkeypatch.setattr(pb, "_enforcer_registered", lambda hook_name: True) + + with caplog.at_level("WARNING"): + result = governed_persist("memory", str(target), b"malformed-enforcer-allow") + + assert result.staged is True + assert result.denied is False + assert not target.exists() + assert "enforcer allow without staged" in caplog.text + + def test_enforcer_present_well_formed_staged_allow_is_unaffected(self, tmp_path, monkeypatch): + # staged=True short-circuits before the enforcer-presence check runs + # at all -- an enforcer that behaves correctly is unaffected by this + # fix either way. + target = tmp_path / "MEMORY.md" + digest = "sha256:" + "b" * 64 + monkeypatch.setattr( + "hermes_cli.plugins.get_required_hook_directive", + _directive("allow", staged=True, digest=digest), + ) + monkeypatch.setattr(pb, "_enforcer_registered", lambda hook_name: True) + + result = governed_persist("memory", str(target), b"content") + + assert result == PersistResult(staged=True, digest=digest, denied=False, message="") + assert not target.exists() + + def test_no_enforcer_bare_allow_is_still_passthrough(self, tmp_path, monkeypatch): + # Explicit no-enforcer case, decoupled from the real plugin manager's + # test-time default -- pins the other half of the fix: nothing + # regresses when there genuinely is no enforcer registered. + target = tmp_path / "MEMORY.md" + monkeypatch.setattr("hermes_cli.plugins.get_required_hook_directive", _directive("allow")) + monkeypatch.setattr(pb, "_enforcer_registered", lambda hook_name: False) + + result = governed_persist("memory", str(target), "hello") + + assert result == PersistResult(staged=False, digest=None, denied=False, message="") + assert target.read_text(encoding="utf-8") == "hello" + + def test_enforcer_registered_reads_the_required_hooks_registry(self, monkeypatch): + class _FakeManager: + _required_hooks = {"pre_persist_write": [("guard", lambda **kw: None)]} + + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: _FakeManager()) + + assert pb._enforcer_registered("pre_persist_write") is True + assert pb._enforcer_registered("pre_run_start") is False + + def test_enforcer_registered_fails_closed_on_introspection_error(self, monkeypatch): + def boom(): + raise RuntimeError("plugin manager unavailable") + + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", boom) + + assert pb._enforcer_registered("pre_persist_write") is True + + # --------------------------------------------------------------------------- # governed_persist -- hook unreachable / unrecognized -> decision-less local stage # --------------------------------------------------------------------------- @@ -407,8 +484,56 @@ def fake_governed_persist(kind, path, content, meta=None): "tempfile.mkstemp(", "atomic_replace(", ".write_text(", + ".write_bytes(", + "os.replace(", + "os.rename(", + "shutil.", ) +# `open(path, "w")` / `Path(...).open("wb")` / `open(path, mode="a")` don't +# contain any fixed token above -- the mode string can land in any argument +# position or as a `mode=` keyword. Caught separately via AST below instead +# of trying to enumerate every quoting/spacing variant as a substring. +_WRITE_MODE_CHARS = frozenset("wax+") + + +def _open_write_mode_findings(source: str, label: str) -> list[str]: + """AST-scan *source* (a standalone function segment or a whole module) + for `open(...)` / `.open(...)` calls whose mode argument permits + writing (contains any of w/a/x/+). A no-mode `open(path)` defaults to + read-only text mode and is not flagged. + """ + tree = ast.parse(source) + findings: list[str] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + is_open_call = (isinstance(func, ast.Name) and func.id == "open") or ( + isinstance(func, ast.Attribute) and func.attr == "open" + ) + if not is_open_call: + continue + mode_value = None + for kw in node.keywords: + if ( + kw.arg == "mode" + and isinstance(kw.value, ast.Constant) + and isinstance(kw.value.value, str) + ): + mode_value = kw.value.value + break + if mode_value is None: + for arg in node.args: + if isinstance(arg, ast.Constant) and isinstance(arg.value, str): + mode_value = arg.value + break + if mode_value is None: + continue + if _WRITE_MODE_CHARS & set(mode_value): + findings.append(f"{label}: open(...) call with write-capable mode {mode_value!r}") + return findings + _WRITE_SITE_FUNCTIONS = ( ("tools/memory_tool.py", "_write_file"), ("tools/skill_manager_tool.py", "_atomic_write_text"), @@ -450,6 +575,12 @@ def test_write_site_functions_contain_no_direct_write_primitive(self): f"write primitive ({snippet!r}) -- route it through " f"agent.persist_boundary.governed_persist instead" ) + open_findings = _open_write_mode_findings(segment, f"{rel}::{func_name}") + assert not open_findings, ( + f"{rel}::{func_name} still contains a direct write-mode " + f"open() call ({open_findings}) -- route it through " + f"agent.persist_boundary.governed_persist instead" + ) def test_delegating_modules_contain_no_direct_write_primitive(self): repo_root = self._repo_root() @@ -461,6 +592,12 @@ def test_delegating_modules_contain_no_direct_write_primitive(self): f"({snippet!r}) -- route it through agent.persist_boundary." f"governed_persist instead" ) + open_findings = _open_write_mode_findings(source, rel) + assert not open_findings, ( + f"{rel} contains a direct write-mode open() call " + f"({open_findings}) -- route it through agent.persist_boundary." + f"governed_persist instead" + ) def test_funnel_module_is_the_one_allowed_writer(self): # Sanity check the allowlist isn't hiding a funnel that silently lost From 32ff8669aee7a02f4fc777b582ab534e2c52a31e Mon Sep 17 00:00:00 2001 From: CN-CLI Date: Mon, 20 Jul 2026 06:40:48 -0600 Subject: [PATCH 3/3] test(agent): AST write-scan reads open() mode from args[1] --- tests/agent/test_persist_boundary.py | 83 ++++++++++++++++++++++++++-- 1 file changed, 77 insertions(+), 6 deletions(-) diff --git a/tests/agent/test_persist_boundary.py b/tests/agent/test_persist_boundary.py index 031382724d70..843bf652a07d 100644 --- a/tests/agent/test_persist_boundary.py +++ b/tests/agent/test_persist_boundary.py @@ -466,6 +466,70 @@ def fake_governed_persist(kind, path, content, meta=None): assert seen["path"].endswith("SKILL.md") +# --------------------------------------------------------------------------- +# Regression tests for _open_write_mode_findings AST helper +# --------------------------------------------------------------------------- + + +class TestOpenWriteModeFindings: + """Unit tests for the _open_write_mode_findings helper to ensure it reads + the correct argument index for function vs. method calls.""" + + def test_open_function_with_write_mode_positional_arg_is_caught(self): + """open("MEMORY.md", "w") should be flagged (mode is args[1], not args[0]).""" + source = ''' +def _write_file(path: str, content: str): + with open("MEMORY.md", "w") as f: + f.write(content) +''' + findings = _open_write_mode_findings(source, "test") + assert len(findings) == 1 + assert "w" in findings[0] + + def test_open_function_with_read_mode_positional_arg_not_flagged(self): + """open("data.xlsx", "r") should NOT be flagged (mode "r" has no write chars).""" + source = ''' +def read_spreadsheet(path: str): + with open("data.xlsx", "r") as f: + return f.read() +''' + findings = _open_write_mode_findings(source, "test") + assert len(findings) == 0 + + def test_path_method_with_write_mode_positional_arg_is_caught(self): + """Path(...).open("w") should be flagged (mode is args[0] for method calls).""" + source = ''' +from pathlib import Path +def write_skill(path: Path, content: str): + path.open("w").write(content) +''' + findings = _open_write_mode_findings(source, "test") + assert len(findings) == 1 + assert "w" in findings[0] + + def test_open_function_with_keyword_mode_is_caught(self): + """open(path, mode="w") with keyword argument should be flagged.""" + source = ''' +def write_file(path: str, content: str): + with open(path, mode="w") as f: + f.write(content) +''' + findings = _open_write_mode_findings(source, "test") + assert len(findings) == 1 + assert "w" in findings[0] + + def test_open_with_append_mode_is_caught(self): + """open(path, "a") with append mode should be flagged.""" + source = ''' +def append_log(path: str, entry: str): + with open(path, "a") as f: + f.write(entry) +''' + findings = _open_write_mode_findings(source, "test") + assert len(findings) == 1 + assert "a" in findings[0] + + # --------------------------------------------------------------------------- # Grep-pin: only the funnel module may perform a canonical write # @@ -509,12 +573,13 @@ def _open_write_mode_findings(source: str, label: str) -> list[str]: if not isinstance(node, ast.Call): continue func = node.func - is_open_call = (isinstance(func, ast.Name) and func.id == "open") or ( - isinstance(func, ast.Attribute) and func.attr == "open" - ) - if not is_open_call: + is_method_call = isinstance(func, ast.Attribute) and func.attr == "open" + is_function_call = isinstance(func, ast.Name) and func.id == "open" + if not (is_method_call or is_function_call): continue + mode_value = None + # Check keyword argument first for kw in node.keywords: if ( kw.arg == "mode" @@ -523,11 +588,17 @@ def _open_write_mode_findings(source: str, label: str) -> list[str]: ): mode_value = kw.value.value break + + # If no keyword mode, check positional arguments if mode_value is None: - for arg in node.args: + # For method calls like Path(...).open("w"), mode is args[0] + # For function calls like open(path, "w"), mode is args[1] + arg_index = 0 if is_method_call else 1 + if arg_index < len(node.args): + arg = node.args[arg_index] if isinstance(arg, ast.Constant) and isinstance(arg.value, str): mode_value = arg.value - break + if mode_value is None: continue if _WRITE_MODE_CHARS & set(mode_value):