From b02d8eb68f80821b8a7fbcccfcbb641ac244e31a Mon Sep 17 00:00:00 2001 From: Violin Date: Thu, 16 Jul 2026 18:23:46 +0100 Subject: [PATCH] refactor: reorganize violin guard core, unify locking and json storage, optimize facades --- plugins/violin_guard/core/__init__.py | 2 + plugins/violin_guard/core/execution.py | 17 +--- plugins/violin_guard/core/release.py | 1 + plugins/violin_guard/core/service.py | 77 +++++++-------- plugins/violin_guard/core/state.py | 124 ++----------------------- plugins/violin_guard/core/storage.py | 109 ++++++++++++++++++++++ 6 files changed, 164 insertions(+), 166 deletions(-) create mode 100644 plugins/violin_guard/core/storage.py diff --git a/plugins/violin_guard/core/__init__.py b/plugins/violin_guard/core/__init__.py index 044e39a..e362d43 100644 --- a/plugins/violin_guard/core/__init__.py +++ b/plugins/violin_guard/core/__init__.py @@ -18,6 +18,7 @@ from . import ( results, service, state, + storage, ) __all__ = [ @@ -32,4 +33,5 @@ __all__ = [ "results", "service", "state", + "storage", ] diff --git a/plugins/violin_guard/core/execution.py b/plugins/violin_guard/core/execution.py index f924e3f..dc6ecd9 100644 --- a/plugins/violin_guard/core/execution.py +++ b/plugins/violin_guard/core/execution.py @@ -6,7 +6,6 @@ This is the ONLY module in core/ that uses subprocess. All others are pure. from __future__ import annotations import contextlib -import json import os import re import shutil @@ -21,6 +20,8 @@ from typing import Any from . import state from .phases import normalize_phase +from .storage import atomic_json as _atomic_json +from .storage import read_json as _read_json __all__ = [ "execute", @@ -47,20 +48,6 @@ def _utc_now() -> str: return datetime.now(UTC).isoformat().replace("+00:00", "Z") -def _atomic_json(path: Path, value: dict[str, Any]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - tmp = path.with_suffix(path.suffix + ".tmp") - tmp.write_text(json.dumps(value, indent=2, sort_keys=True), encoding="utf-8") - tmp.replace(path) - - -def _read_json(path: Path) -> dict[str, Any]: - try: - return json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return {} - - def _resolve_engagement(eng_dir: str) -> Path: path = state._eng_dir(eng_dir) if not path.is_dir(): diff --git a/plugins/violin_guard/core/release.py b/plugins/violin_guard/core/release.py index 508297e..8ed0dce 100644 --- a/plugins/violin_guard/core/release.py +++ b/plugins/violin_guard/core/release.py @@ -258,6 +258,7 @@ def validate_plugin_structure() -> StructureResult: "adapters.py", "bootstrap.py", "release.py", + "storage.py", ] for mod in core_modules: if (core_dir / mod).exists(): diff --git a/plugins/violin_guard/core/service.py b/plugins/violin_guard/core/service.py index 6f349d6..5ee58d3 100644 --- a/plugins/violin_guard/core/service.py +++ b/plugins/violin_guard/core/service.py @@ -23,8 +23,8 @@ def _result(r): return {"errors": r.errors, "warnings": r.warnings, "infos": r.infos} -def handle_check_command(a, **kwargs): - r = command.check_command( +def _check_command_internal(a) -> command.CheckResult: + return command.check_command( command.CheckCommandArgs( command=a.get("command", ""), phase=a.get("phase", ""), @@ -35,9 +35,12 @@ def handle_check_command(a, **kwargs): skill_loaded_file=a.get("skill_loaded_file"), ) ) - return _json( - "ok" if r.exit_code() == 0 else "review" if r.exit_code() == 2 else "block", **_result(r) - ) + + +def handle_check_command(a, **kwargs): + r = _check_command_internal(a) + status_name = "ok" if r.exit_code() == 0 else "review" if r.exit_code() == 2 else "block" + return _json(status_name, **_result(r)) def handle_record_ptt(a, **kwargs): @@ -276,23 +279,25 @@ def handle_heartbeat_done(a, **kwargs): def handle_exec(a, **kwargs): - gate = json.loads(handle_check_command(a)) - if gate["status"] not in ("ok",) and not ( - gate["status"] == "review" and os.environ.get("HERMES_YOLO_MODE") == "1" + r = _check_command_internal(a) + exit_code = r.exit_code() + status_name = "ok" if exit_code == 0 else "review" if exit_code == 2 else "block" + if status_name not in ("ok",) and not ( + status_name == "review" and os.environ.get("HERMES_YOLO_MODE") == "1" ): status = ( "sync_required" if any( - "sync-credit" in str(x) or "not synced" in str(x) for x in gate.get("errors", []) + "sync-credit" in str(x) or "not synced" in str(x) for x in r.errors ) else "denied" ) - return _json(status, executed=False, **gate) + return _json(status, executed=False, **_result(r)) try: active_task = ptt.find_active_task( ptt.parse_ptt(_eng_path(a["eng_dir"]) / "state" / "ptt.md") ) - r = execution.execute( + res = execution.execute( command=a["command"], eng_dir=a["eng_dir"], phase=a["phase"], @@ -304,8 +309,8 @@ def handle_exec(a, **kwargs): argv=a.get("_argv"), background=bool(a.get("background", False)), ) - r.pop("status", None) - return _json("ok", **r) + res.pop("status", None) + return _json("ok", **res) except Exception as e: return _json("execution_failed", error=str(e), executed=False) @@ -356,24 +361,22 @@ def handle_exec_burst(a, **kwargs): active_task = ptt.find_active_task(ptt.parse_ptt(_eng_path(eng_dir) / "state" / "ptt.md")) active_task_id = active_task.id if active_task else "" - results = [] executed = 0 for idx, cmd in enumerate(cmds): - gate = json.loads( - handle_check_command( - { - "command": cmd, - "phase": phase, - "eng_dir": eng_dir, - "scope": scope, - "session_id": session_id, - "skill_loaded_file": skill_loaded_file, - "target": a.get("target"), - } - ) - ) - if gate["status"] == "block": + cmd_args = { + "command": cmd, + "phase": phase, + "eng_dir": eng_dir, + "scope": scope, + "session_id": session_id, + "skill_loaded_file": skill_loaded_file, + "target": a.get("target"), + } + r = _check_command_internal(cmd_args) + exit_code = r.exit_code() + status_name = "ok" if exit_code == 0 else "review" if exit_code == 2 else "block" + if status_name == "block": # Hard block — never continue; halt the batch fail-closed. return _json( "denied", @@ -384,12 +387,12 @@ def handle_exec_burst(a, **kwargs): "index": idx + 1, "command": cmd, "status": "blocked", - "errors": gate.get("errors", []), + "errors": r.errors, } ], - reason=f"command [{idx + 1}] blocked: {gate.get('errors', ['blocked'])[0]}", + reason=f"command [{idx + 1}] blocked: {r.errors[0] if r.errors else 'blocked'}", ) - if gate["status"] == "review" and os.environ.get("HERMES_YOLO_MODE") != "1": + if status_name == "review" and os.environ.get("HERMES_YOLO_MODE") != "1": # Soft review blocks unless yolo overrides; also halts the batch. return _json( "denied", @@ -400,13 +403,13 @@ def handle_exec_burst(a, **kwargs): "index": idx + 1, "command": cmd, "status": "review_required", - "warnings": gate.get("warnings", []), + "warnings": r.warnings, } ], reason=f"command [{idx + 1}] requires review before execution", ) try: - r = execution.execute( + res = execution.execute( command=cmd, eng_dir=eng_dir, phase=phase, @@ -416,13 +419,13 @@ def handle_exec_burst(a, **kwargs): label=label, ptt_task_id=active_task_id, ) - r.pop("status", None) - entry = {"index": idx + 1, "command": cmd, **r} + res.pop("status", None) + entry = {"index": idx + 1, "command": cmd, **res} results.append(entry) - if r.get("executed"): + if res.get("executed"): executed += 1 # A target command that ran but failed: honor continue_on_error. - if r.get("exit_code", 0) != 0 and not continue_on_error: + if res.get("exit_code", 0) != 0 and not continue_on_error: break except Exception as e: # noqa: BLE001 - executor error must not abort silently if not continue_on_error: diff --git a/plugins/violin_guard/core/state.py b/plugins/violin_guard/core/state.py index 32c9f33..d54aa95 100644 --- a/plugins/violin_guard/core/state.py +++ b/plugins/violin_guard/core/state.py @@ -5,24 +5,21 @@ Pure functions with atomic, cross-process-locked file operations. No subprocess from __future__ import annotations -import contextlib -import json import os -import time from datetime import UTC, datetime from pathlib import Path from typing import Any -try: # POSIX - import fcntl -except ImportError: # Windows - fcntl = None -try: # Windows - import msvcrt -except ImportError: # POSIX - msvcrt = None - -from .phases import Phase +from .phases import suppresses_heartbeat +from .storage import ( + lock_file as _lock_file, +) +from .storage import ( + mutate_json as _mutate_json, +) +from .storage import ( + read_json as _read_json, +) # Constants DEFAULT_SYNC_CREDIT = 5 @@ -76,102 +73,6 @@ def _state_dir(eng_dir: str | Path) -> Path: return p -def _lock_file(path: Path): - """Acquire an exclusive advisory lock for the duration of a ``with`` block. - - Uses ``fcntl`` on POSIX and ``msvcrt`` on Windows. The lock is held on the - target file's directory lockfile (named ``.lock``) so concurrent - processes serialise writes without racing on the temp swap. - """ - lock_path = path.with_suffix(path.suffix + ".lock") - lock_path.parent.mkdir(parents=True, exist_ok=True) - # ``msvcrt.locking`` locks bytes, so the file must contain at least one. - # Opening in binary append mode also avoids truncating a lock file another - # process has already opened. - fh = open(lock_path, "a+b") # noqa: SIM115 - closed in _FileLock - if msvcrt is not None: - fh.seek(0, 2) - if fh.tell() == 0: - fh.write(b"0") - fh.flush() - fh.seek(0) - if fcntl is not None: - try: - fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - except OSError: - # Blocking fallback: wait for the lock to free. - fcntl.flock(fh.fileno(), fcntl.LOCK_EX) - elif msvcrt is not None: - # msvcrt has no non-blocking mode; retry briefly. - deadline = time.monotonic() + 5.0 - while True: - try: - msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1) - break - except OSError as exc: - if time.monotonic() >= deadline: - fh.close() - raise TimeoutError(f"timed out acquiring state lock: {lock_path}") from exc - time.sleep(0.05) - return _FileLock(fh) - - -class _FileLock: - def __init__(self, fh): - self._fh = fh - - def __enter__(self): - return self - - def __exit__(self, *exc): - if self._fh is None: - return False - try: - if fcntl is not None: - fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN) - elif msvcrt is not None: - with contextlib.suppress(OSError): - msvcrt.locking(self._fh.fileno(), msvcrt.LK_UNLCK, 1) - finally: - self._fh.close() - return False - - -def _atomic_write_locked(path: Path, data: dict[str, Any]) -> None: - """Write JSON atomically while the caller holds ``path``'s lock.""" - path.parent.mkdir(parents=True, exist_ok=True) - tmp = path.with_suffix(path.suffix + ".tmp") - tmp.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8") - tmp.replace(path) - - -def _atomic_write(path: Path, data: dict[str, Any]) -> None: - """Atomic JSON write using tmp + os.replace, guarded by an advisory lock.""" - with _lock_file(path): - _atomic_write_locked(path, data) - - -def _read_json(path: Path) -> dict[str, Any]: - try: - return json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return {} - - -def _mutate_json(path: Path, mutation): - """Apply ``mutation`` to one state document under a single file lock. - - Every state transition must read, modify and replace the document while - holding the same lock. Locking only the final replace loses updates under - concurrent tool calls. - """ - with _lock_file(path): - data = _read_json(path) - result = mutation(data) - _atomic_write_locked(path, data) - return result - - # --------------------------------------------------------------------------- # # Sync credit / pending sync # --------------------------------------------------------------------------- # @@ -475,11 +376,6 @@ def artifacts_are_fresh(eng_dir: str | Path) -> bool: return all(p.exists() for p in paths) -def suppresses_heartbeat(phase: Phase) -> bool: - """Return True for phases that suppress heartbeat (EXPLOITATION, POST_EXPLOITATION).""" - return phase in (Phase.EXPLOITATION, Phase.POST_EXPLOITATION) - - __all__ = [ "DEFAULT_SYNC_CREDIT", "COMMAND_INTERVAL", diff --git a/plugins/violin_guard/core/storage.py b/plugins/violin_guard/core/storage.py new file mode 100644 index 0000000..fa8b1b1 --- /dev/null +++ b/plugins/violin_guard/core/storage.py @@ -0,0 +1,109 @@ +"""Unified file storage, locking, and JSON serialization utilities.""" + +from __future__ import annotations + +import contextlib +import json +import time +from pathlib import Path +from typing import Any + +try: # POSIX + import fcntl +except ImportError: # Windows + fcntl = None +try: # Windows + import msvcrt +except ImportError: # POSIX + msvcrt = None + + +def lock_file(path: Path) -> FileLock: + """Acquire an exclusive advisory lock for the duration of a ``with`` block. + + Uses ``fcntl`` on POSIX and ``msvcrt`` on Windows. The lock is held on the + target file's directory lockfile (named ``.lock``) so concurrent + processes serialise writes without racing on the temp swap. + """ + lock_path = path.with_suffix(path.suffix + ".lock") + lock_path.parent.mkdir(parents=True, exist_ok=True) + # ``msvcrt.locking`` locks bytes, so the file must contain at least one. + # Opening in binary append mode also avoids truncating a lock file another + # process has already opened. + fh = open(lock_path, "a+b") # noqa: SIM115 - closed in FileLock + if msvcrt is not None: + fh.seek(0, 2) + if fh.tell() == 0: + fh.write(b"0") + fh.flush() + fh.seek(0) + if fcntl is not None: + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + # Blocking fallback: wait for the lock to free. + fcntl.flock(fh.fileno(), fcntl.LOCK_EX) + elif msvcrt is not None: + # msvcrt has no non-blocking mode; retry briefly. + deadline = time.monotonic() + 5.0 + while True: + try: + msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1) + break + except OSError as exc: + if time.monotonic() >= deadline: + fh.close() + raise TimeoutError(f"timed out acquiring state lock: {lock_path}") from exc + time.sleep(0.05) + return FileLock(fh) + + +class FileLock: + def __init__(self, fh): + self._fh = fh + + def __enter__(self): + return self + + def __exit__(self, *exc): + if self._fh is None: + return False + try: + if fcntl is not None: + fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN) + elif msvcrt is not None: + with contextlib.suppress(OSError): + msvcrt.locking(self._fh.fileno(), msvcrt.LK_UNLCK, 1) + finally: + self._fh.close() + return False + + +def read_json(path: Path) -> dict[str, Any]: + """Read a JSON document, returning an empty dict on error.""" + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + + +def atomic_json(path: Path, data: dict[str, Any]) -> None: + """Write JSON atomically by replacing a temporary swap file.""" + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8") + tmp.replace(path) + + +def mutate_json(path: Path, mutation) -> Any: + """Apply ``mutation`` to one state document under a single file lock. + + Every state transition must read, modify and replace the document while + holding the same lock. Locking only the final replace loses updates under + concurrent tool calls. + """ + with lock_file(path): + data = read_json(path) + result = mutation(data) + atomic_json(path, data) + return result