External/violin
1
0
Fork 0
mirror of https://github.com/Strategic-Automation/violin.git synced 2026-08-14 12:33:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: reorganize violin guard core, unify locking and json storage, optimize facades

Browse Source
This commit is contained in:
Violin
2026-07-16 18:23:46 +01:00
parent 56387e2577
commit b02d8eb68f
6 changed files with 164 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/violin_guard/core/__init__.py
+2
View File
@@ -18,6 +18,7 @@ from . import (
results,
service,
state,
storage,
)
__all__ = [
@@ -32,4 +33,5 @@ __all__ = [
"results",
"service",
"state",
"storage",
]
plugins/violin_guard/core/execution.py
+2 -15
View File
@@ -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():
plugins/violin_guard/core/release.py
+1
View File
@@ -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():
plugins/violin_guard/core/service.py
+40 -37
View File
@@ -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:
plugins/violin_guard/core/state.py
+10 -114
View File
@@ -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 ``<file>.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",
plugins/violin_guard/core/storage.py
+109
View File
@@ -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 ``<file>.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