mirror of
https://github.com/Strategic-Automation/violin.git
synced 2026-08-14 12:33:37 +02:00
630 lines
21 KiB
Python
630 lines
21 KiB
Python
"""State machine, advisory file locking, and JSON storage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import time
|
|
import uuid
|
|
from contextlib import contextmanager
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from filelock import FileLock
|
|
|
|
# Constants
|
|
|
|
DEFAULT_SYNC_CREDIT = 5
|
|
COMMAND_INTERVAL = 50
|
|
MAX_BURST_COMMANDS = 20
|
|
PHASE_SYNC_CREDIT = {
|
|
"RECON": 10,
|
|
"VULN_RESEARCH": 10,
|
|
"EXPLOITATION": 20,
|
|
"POST_EXPLOITATION": 20,
|
|
"PRIVESC": 20,
|
|
"FLAGS": 20,
|
|
}
|
|
|
|
# Local tools
|
|
LOCAL_TOOLS = {"echo", "true", "false", "printf", "pwd", "ls", "cat", "date"}
|
|
|
|
_STATE_DIR = "state"
|
|
_SYNC_FILE = "sync.json"
|
|
_HEARTBEAT_FILE = "heartbeat.json"
|
|
_COUNTS_FILE = "counts.json"
|
|
_SESSION_FILE = "session.json"
|
|
_SEMANTIC_FILE = "semantic-progress.json"
|
|
|
|
|
|
# Path helpers
|
|
|
|
|
|
def _eng_root() -> Path:
|
|
"""Return Violin's stable profile/repository root for relative paths."""
|
|
override = os.environ.get("VIOLIN_ENG_ROOT", "").strip()
|
|
if override:
|
|
return Path(override).expanduser().resolve()
|
|
container_root = Path("/violin")
|
|
if container_root.exists() and (container_root / "engagements").exists():
|
|
return container_root.resolve()
|
|
return Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def resolve_eng_dir(eng_dir: str | Path) -> Path:
|
|
"""Resolve an engagement directory path (absolute or relative to profile root)."""
|
|
if not str(eng_dir).strip() or str(eng_dir).strip() == ".":
|
|
env_eng = os.environ.get("ENG_DIR", "").strip()
|
|
if env_eng:
|
|
return Path(env_eng).expanduser().resolve()
|
|
cwd = Path.cwd().resolve()
|
|
if (cwd / "scope" / "scope.yaml").exists() or (cwd / "hypotheses.md").exists():
|
|
return cwd
|
|
return _eng_root()
|
|
|
|
path = Path(eng_dir).expanduser()
|
|
if not path.is_absolute():
|
|
profile_candidate = (_eng_root() / path).resolve()
|
|
cwd_candidate = (Path.cwd() / path).resolve()
|
|
if not profile_candidate.exists() and cwd_candidate.exists():
|
|
return cwd_candidate
|
|
return profile_candidate
|
|
return path.resolve()
|
|
|
|
|
|
def resolve_session_id(eng_dir: str | Path, session_id: str | None = None) -> str:
|
|
"""Return an explicit session id or the engagement's recorded session.
|
|
|
|
Tool calls should not fail merely because the runtime omitted a value it
|
|
already supplied to the lifecycle hook. Older engagements are supported
|
|
by inferring the id when they contain exactly one skill-load marker.
|
|
"""
|
|
if session_id and session_id.strip():
|
|
return session_id.strip()
|
|
root = resolve_eng_dir(eng_dir)
|
|
recorded = str(read_json(root / _STATE_DIR / _SESSION_FILE).get("session_id") or "").strip()
|
|
if recorded:
|
|
return recorded
|
|
markers = (
|
|
list((root / _STATE_DIR).glob(".skill-loaded-*")) if (root / _STATE_DIR).exists() else []
|
|
)
|
|
return markers[0].name.removeprefix(".skill-loaded-") if len(markers) == 1 else ""
|
|
|
|
|
|
def record_session_id(eng_dir: str | Path, session_id: str | None) -> None:
|
|
if session_id and session_id.strip():
|
|
path = _state_dir(eng_dir) / _SESSION_FILE
|
|
with lock_file(path):
|
|
atomic_json(path, {"session_id": session_id.strip()})
|
|
|
|
|
|
def ensure_dir(path: Path) -> Path:
|
|
"""Ensure directory exists fail-safe against symlinks and cross-platform FileExistsError [Errno 17]."""
|
|
try:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
except FileExistsError:
|
|
if not path.exists():
|
|
raise
|
|
return path
|
|
|
|
|
|
def _state_dir(eng_dir: str | Path) -> Path:
|
|
p = resolve_eng_dir(eng_dir) / _STATE_DIR
|
|
ensure_dir(p)
|
|
return p
|
|
|
|
|
|
# Storage primitives
|
|
|
|
|
|
@contextmanager
|
|
def lock_file(path: Path):
|
|
"""Acquire an exclusive advisory lock on ``path`` for the duration of a ``with`` block."""
|
|
lock_path = path.with_suffix(path.suffix + ".lock")
|
|
ensure_dir(lock_path.parent)
|
|
with FileLock(str(lock_path), timeout=20):
|
|
yield
|
|
|
|
|
|
@contextmanager
|
|
def workflow_lock(eng_dir: str | Path):
|
|
"""Serialize multi-file workflow transitions for one engagement.
|
|
|
|
Callers acquire this lock before any narrower JSON or receipt file lock.
|
|
"""
|
|
lock_path = _state_dir(eng_dir) / "workflow.lock"
|
|
with FileLock(str(lock_path), timeout=20):
|
|
yield
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
|
"""Read a JSON document, returning an empty dict on missing file or non-dict root.
|
|
|
|
Raises OSError or json.JSONDecodeError on corrupt/locked file reads when the file exists,
|
|
preventing mutate_json from overwriting existing state with empty dictionaries.
|
|
"""
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except (OSError, json.JSONDecodeError):
|
|
# On read failure when file exists, attempt up to 3 retries for transient locks
|
|
for attempt in range(3):
|
|
time.sleep(0.02 * (attempt + 1))
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
raise
|
|
|
|
|
|
def atomic_json(path: Path, data: dict[str, Any]) -> None:
|
|
"""Write JSON atomically by replacing a temporary swap file."""
|
|
ensure_dir(path.parent)
|
|
tmp = path.with_name(f"{path.name}.{uuid.uuid4().hex}.tmp")
|
|
tmp.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8")
|
|
try:
|
|
for attempt in range(5):
|
|
try:
|
|
tmp.replace(path)
|
|
return
|
|
except PermissionError:
|
|
if attempt == 4:
|
|
raise
|
|
time.sleep(0.02 * (attempt + 1))
|
|
finally:
|
|
if tmp.exists():
|
|
with contextlib.suppress(OSError):
|
|
tmp.unlink()
|
|
|
|
|
|
def atomic_text(path: Path, content: str) -> None:
|
|
"""Write one UTF-8 text document with an atomic replace."""
|
|
ensure_dir(path.parent)
|
|
tmp = path.with_name(f"{path.name}.{uuid.uuid4().hex}.tmp")
|
|
tmp.write_text(content, encoding="utf-8")
|
|
try:
|
|
for attempt in range(5):
|
|
try:
|
|
tmp.replace(path)
|
|
return
|
|
except PermissionError:
|
|
if attempt == 4:
|
|
raise
|
|
time.sleep(0.02 * (attempt + 1))
|
|
finally:
|
|
if tmp.exists():
|
|
with contextlib.suppress(OSError):
|
|
tmp.unlink()
|
|
|
|
|
|
def mutate_json(path: Path, mutation) -> Any:
|
|
"""Apply ``mutation`` to one state document under a single file lock."""
|
|
with lock_file(path):
|
|
data = read_json(path)
|
|
result = mutation(data)
|
|
atomic_json(path, data)
|
|
return result
|
|
|
|
|
|
# Local command classification
|
|
|
|
|
|
def is_local_bookkeeping_command(command: str) -> bool:
|
|
"""Whether a command is a harmless local bookkeeping action."""
|
|
from .bash_ast import parse_bash_segments
|
|
from .targets import extract_target_candidates
|
|
|
|
segments = parse_bash_segments(command)
|
|
if len(segments) != 1:
|
|
return False
|
|
segment = segments[0]
|
|
if segment.executable not in LOCAL_TOOLS or segment.redirects:
|
|
return False
|
|
return not extract_target_candidates(command)
|
|
|
|
|
|
# Sync credit / pending sync
|
|
|
|
|
|
def _sync_path(eng_dir: str | Path) -> Path:
|
|
return _state_dir(eng_dir) / _SYNC_FILE
|
|
|
|
|
|
def sync_credit_limit(phase: str | None = None) -> int:
|
|
key = str(phase or "").strip().upper().replace("-", "_")
|
|
return PHASE_SYNC_CREDIT.get(key, DEFAULT_SYNC_CREDIT)
|
|
|
|
|
|
def sync_credit_remaining(eng_dir: str | Path, phase: str | None = None) -> int:
|
|
data = read_json(_sync_path(eng_dir))
|
|
return max(0, data.get("credit", sync_credit_limit(phase)))
|
|
|
|
|
|
def spend_sync_credit(eng_dir: str | Path, phase: str) -> int:
|
|
path = _sync_path(eng_dir)
|
|
|
|
def spend(data: dict[str, Any]) -> int:
|
|
starting_credit = data.get("credit", sync_credit_limit(phase))
|
|
credit = max(0, starting_credit - 1)
|
|
data["credit"] = credit
|
|
return credit
|
|
|
|
return mutate_json(path, spend)
|
|
|
|
|
|
def reserve_sync_credit(eng_dir: str | Path, phase: str, count: int) -> str:
|
|
"""Atomically reserve credit for a burst before any command starts."""
|
|
if count < 1:
|
|
raise ValueError("a sync reservation must contain at least one command")
|
|
path = _sync_path(eng_dir)
|
|
|
|
def reserve(data: dict[str, Any]) -> str:
|
|
credit = max(0, int(data.get("credit", sync_credit_limit(phase))))
|
|
if credit < count:
|
|
raise ValueError(f"insufficient sync credit for burst: need {count}, have {credit}")
|
|
reservation_id = f"burst-{uuid.uuid4().hex}"
|
|
data["credit"] = credit - count
|
|
data.setdefault("reservations", {})[reservation_id] = {
|
|
"phase": phase,
|
|
"remaining": count,
|
|
"created_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
|
}
|
|
return reservation_id
|
|
|
|
return mutate_json(path, reserve)
|
|
|
|
|
|
def consume_reserved_sync_credit(eng_dir: str | Path, reservation_id: str) -> int:
|
|
"""Consume one previously reserved slot without decrementing credit twice."""
|
|
path = _sync_path(eng_dir)
|
|
|
|
def consume(data: dict[str, Any]) -> int:
|
|
reservation = (data.get("reservations") or {}).get(reservation_id)
|
|
if not reservation or int(reservation.get("remaining", 0)) < 1:
|
|
raise ValueError("sync reservation is missing or exhausted")
|
|
reservation["remaining"] = int(reservation["remaining"]) - 1
|
|
if reservation["remaining"] == 0:
|
|
data["reservations"].pop(reservation_id, None)
|
|
return max(0, int(data.get("credit", 0)))
|
|
|
|
return mutate_json(path, consume)
|
|
|
|
|
|
def release_reserved_sync_credit(eng_dir: str | Path, reservation_id: str) -> int:
|
|
"""Return every unconsumed slot in a reservation to the sync window."""
|
|
path = _sync_path(eng_dir)
|
|
|
|
def release(data: dict[str, Any]) -> int:
|
|
reservation = (data.get("reservations") or {}).pop(reservation_id, None)
|
|
if reservation:
|
|
data["credit"] = int(data.get("credit", 0)) + max(
|
|
0, int(reservation.get("remaining", 0))
|
|
)
|
|
return max(0, int(data.get("credit", 0)))
|
|
|
|
return mutate_json(path, release)
|
|
|
|
|
|
def mark_pending_sync(
|
|
eng_dir: str | Path,
|
|
command: str,
|
|
command_phase: str,
|
|
ptt_task_id: str,
|
|
) -> None:
|
|
path = _sync_path(eng_dir)
|
|
|
|
def mark(data: dict[str, Any]) -> None:
|
|
old = data.get("pending") or {}
|
|
commands = list(old.get("commands") or [])
|
|
if old.get("command") and not commands:
|
|
commands = [{"command": old["command"], "phase": old.get("phase", command_phase)}]
|
|
commands.append({"command": command, "phase": command_phase})
|
|
task_id = old.get("ptt_task_id") or ptt_task_id
|
|
if not task_id:
|
|
raise ValueError("pending execution requires a captured active PTT task")
|
|
data["pending"] = {
|
|
"batch_id": old.get("batch_id") or str(uuid.uuid4()),
|
|
"commands": commands,
|
|
"phase": command_phase,
|
|
"created_at": old.get("created_at")
|
|
or datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
|
"ptt_task_id": task_id,
|
|
"ptt_reviewed": False,
|
|
"credit_limit": old.get("credit_limit") or sync_credit_limit(command_phase),
|
|
}
|
|
|
|
mutate_json(path, mark)
|
|
|
|
|
|
def clear_pending_sync(eng_dir: str | Path) -> None:
|
|
path = _sync_path(eng_dir)
|
|
|
|
def clear(data: dict[str, Any]) -> None:
|
|
data.pop("pending", None)
|
|
data.pop("credit", None)
|
|
data.pop("reservations", None)
|
|
|
|
mutate_json(path, clear)
|
|
|
|
|
|
def has_pending_sync(eng_dir: str | Path) -> bool:
|
|
data = read_json(_sync_path(eng_dir))
|
|
return "pending" in data
|
|
|
|
|
|
def get_pending_sync(eng_dir: str | Path) -> dict | None:
|
|
data = read_json(_sync_path(eng_dir))
|
|
return data.get("pending")
|
|
|
|
|
|
def rebind_pending_sync(
|
|
eng_dir: str | Path,
|
|
*,
|
|
expected_batch_id: str,
|
|
current_task_id: str,
|
|
replacement_task_id: str,
|
|
note: str,
|
|
) -> dict[str, Any]:
|
|
"""Rebind a completed pending batch without certifying its PTT review."""
|
|
path = _sync_path(eng_dir)
|
|
|
|
def rebind(data: dict[str, Any]) -> dict[str, Any]:
|
|
pending = data.get("pending")
|
|
if not pending:
|
|
raise ValueError("no pending execution batch")
|
|
batch_id = str(pending.get("batch_id") or "")
|
|
if batch_id != expected_batch_id:
|
|
raise ValueError(
|
|
f"stale batch id {expected_batch_id!r}; current pending batch is {batch_id!r}"
|
|
)
|
|
captured = str(pending.get("ptt_task_id") or "")
|
|
if captured != current_task_id:
|
|
raise ValueError(
|
|
f"current task {current_task_id!r} does not match batch task {captured!r}"
|
|
)
|
|
if current_task_id == replacement_task_id:
|
|
raise ValueError("replacement task must differ from the current batch task")
|
|
|
|
entry = {
|
|
"timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
|
"batch_id": batch_id,
|
|
"old_task_id": current_task_id,
|
|
"new_task_id": replacement_task_id,
|
|
"note": note.strip(),
|
|
}
|
|
data.setdefault("rebind_audit", []).append(entry)
|
|
pending["ptt_task_id"] = replacement_task_id
|
|
pending["ptt_reviewed"] = False
|
|
pending.pop("ptt_note", None)
|
|
pending.pop("ptt_reviewed_at", None)
|
|
data["pending"] = pending
|
|
return entry
|
|
|
|
return mutate_json(path, rebind)
|
|
|
|
|
|
def mark_ptt_reviewed(eng_dir: str | Path, task_id: str, note: str) -> None:
|
|
path = _sync_path(eng_dir)
|
|
|
|
def mark(data: dict[str, Any]) -> None:
|
|
pending = data.get("pending")
|
|
if not pending:
|
|
raise ValueError("no pending execution batch")
|
|
pending["ptt_reviewed"] = True
|
|
pending["ptt_task_id"] = task_id
|
|
pending["ptt_note"] = note.strip()
|
|
pending["ptt_reviewed_at"] = datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
|
data["pending"] = pending
|
|
|
|
mutate_json(path, mark)
|
|
|
|
|
|
def record_semantic_review(
|
|
eng_dir: str | Path,
|
|
*,
|
|
task_id: str,
|
|
hypothesis_id: str,
|
|
skill: str,
|
|
technique: str,
|
|
outcome: str,
|
|
evidence_paths: list[str],
|
|
next_action: str,
|
|
next_technique: str,
|
|
research_attempted: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Track evidence-backed technique-pivot progress and the anti-stuck lock.
|
|
|
|
The anti-stuck lock is meant to catch *circular* recon — repeating the same
|
|
path without learning. It therefore counts *distinct technique pivots*
|
|
(each ``technique`` key, tracked via ``violin_record_hypothesis`` / the
|
|
``technique`` argument), not raw ``violin_review_batch`` calls. A review
|
|
resets the no-progress counter when it is either:
|
|
|
|
* **evidence-backed** — ``outcome`` is progress/validated/rejected *and*
|
|
the batch carried completed execution evidence, or
|
|
* **a genuine pivot** — ``next_technique`` differs from the current
|
|
``technique`` (a new attack path), which is exactly the behaviour the
|
|
lock exists to encourage.
|
|
|
|
Low-evidence iterative CTF recon therefore never accumulates toward a lock
|
|
as long as each step pivots to a new technique or records evidence.
|
|
"""
|
|
|
|
path = _state_dir(eng_dir) / _SEMANTIC_FILE
|
|
key = "|".join((task_id, hypothesis_id, skill, technique.strip().lower()))
|
|
clean_evidence_paths = [p for p in evidence_paths if p]
|
|
has_evidence = bool(clean_evidence_paths)
|
|
positive = outcome in {"progress", "validated", "rejected"} and has_evidence
|
|
pivoted = bool(
|
|
next_technique.strip().lower()
|
|
and next_technique.strip().lower() != technique.strip().lower()
|
|
)
|
|
|
|
def record(data: dict[str, Any]) -> dict[str, Any]:
|
|
entries = data.setdefault("entries", {})
|
|
entry = entries.get(key, {"count": 0})
|
|
# Reset the per-technique no-progress counter on evidence-backed output
|
|
# or a real pivot; otherwise increment it as a stuck repetition.
|
|
count = 0 if (positive or pivoted) else int(entry.get("count") or 0) + 1
|
|
entry.update(
|
|
{
|
|
"count": count,
|
|
"outcome": outcome,
|
|
"evidence_paths": clean_evidence_paths,
|
|
"next_action": next_action,
|
|
"next_technique": next_technique,
|
|
"pivoted": pivoted,
|
|
"updated_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
|
}
|
|
)
|
|
entries[key] = entry
|
|
lock = data.get("lock") or {}
|
|
# Whole-engagement stuck signal: total no-progress reviews across all
|
|
# keys. Pivots and evidence reset it, so a busy CTF loop stays open.
|
|
total_stuck = sum(
|
|
int(item.get("count") or 0) for item in entries.values() if not item.get("pivoted")
|
|
)
|
|
if lock and (has_evidence or (data.get("research_attempts") and pivoted)):
|
|
data.pop("lock", None)
|
|
elif total_stuck >= 5 and not pivoted and not has_evidence:
|
|
data["lock"] = {
|
|
"key": key,
|
|
"count": total_stuck,
|
|
"reason": "five technique no-progress reviews without a pivot or evidence",
|
|
}
|
|
return {
|
|
"count": count,
|
|
"warning": total_stuck >= 3,
|
|
"locked": bool(data.get("lock")),
|
|
}
|
|
|
|
return mutate_json(path, record)
|
|
|
|
|
|
def record_research_attempt(eng_dir: str | Path, tool_name: str, success: bool) -> None:
|
|
"""Record an actual web research-tool attempt for semantic-lock recovery."""
|
|
|
|
path = _state_dir(eng_dir) / _SEMANTIC_FILE
|
|
|
|
def record(data: dict[str, Any]) -> None:
|
|
attempts = data.setdefault("research_attempts", [])
|
|
attempts.append(
|
|
{
|
|
"tool": tool_name,
|
|
"success": success,
|
|
"timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
|
}
|
|
)
|
|
data["research_attempts"] = attempts[-20:]
|
|
|
|
mutate_json(path, record)
|
|
|
|
|
|
def semantic_lock(eng_dir: str | Path) -> dict[str, Any] | None:
|
|
return read_json(_state_dir(eng_dir) / _SEMANTIC_FILE).get("lock")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heartbeat
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _heartbeat_path(eng_dir: str | Path) -> Path:
|
|
return _state_dir(eng_dir) / _HEARTBEAT_FILE
|
|
|
|
|
|
def set_heartbeat_pending(eng_dir: str | Path, reason: str) -> None:
|
|
path = _heartbeat_path(eng_dir)
|
|
|
|
def mark(data: dict[str, Any]) -> None:
|
|
data["pending"] = True
|
|
data["reason"] = reason
|
|
data["created_at"] = datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
|
|
|
mutate_json(path, mark)
|
|
|
|
|
|
def clear_heartbeat_pending(eng_dir: str | Path) -> None:
|
|
path = _heartbeat_path(eng_dir)
|
|
|
|
def clear(data: dict[str, Any]) -> None:
|
|
data["pending"] = False
|
|
data.pop("reason", None)
|
|
|
|
mutate_json(path, clear)
|
|
|
|
|
|
def has_heartbeat_pending(eng_dir: str | Path) -> bool:
|
|
data = read_json(_heartbeat_path(eng_dir))
|
|
return data.get("pending", False)
|
|
|
|
|
|
def get_heartbeat_reason(eng_dir: str | Path) -> str | None:
|
|
data = read_json(_heartbeat_path(eng_dir))
|
|
return data.get("reason")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command / message counters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _counts_path(eng_dir: str | Path) -> Path:
|
|
return _state_dir(eng_dir) / _COUNTS_FILE
|
|
|
|
|
|
def read_counts(eng_dir: str | Path) -> dict[str, int]:
|
|
data = read_json(_counts_path(eng_dir))
|
|
return {
|
|
"commands": data.get("commands", 0),
|
|
"messages": data.get("messages", 0),
|
|
}
|
|
|
|
|
|
def tick_command(eng_dir: str | Path) -> int:
|
|
path = _counts_path(eng_dir)
|
|
|
|
def tick(data: dict[str, Any]) -> int:
|
|
data["commands"] = data.get("commands", 0) + 1
|
|
return data["commands"]
|
|
|
|
return mutate_json(path, tick)
|
|
|
|
|
|
def tick_message(eng_dir: str | Path) -> int:
|
|
path = _counts_path(eng_dir)
|
|
|
|
def tick(data: dict[str, Any]) -> int:
|
|
data["messages"] = data.get("messages", 0) + 1
|
|
return data["messages"]
|
|
|
|
# Windows can briefly deny the replace/read sequence immediately after a
|
|
# prior hook writes this same file. Lifecycle hooks intentionally do not
|
|
# fail the model turn, so retry here rather than silently dropping a tick.
|
|
for attempt in range(3):
|
|
try:
|
|
return mutate_json(path, tick)
|
|
except OSError:
|
|
if attempt == 2:
|
|
raise
|
|
time.sleep(0.02 * (attempt + 1))
|
|
raise RuntimeError("unreachable")
|
|
|
|
|
|
def record_ok_check(eng_dir: str | Path, command: str, phase: str) -> None:
|
|
path = _counts_path(eng_dir)
|
|
|
|
def record(data: dict[str, Any]) -> None:
|
|
data["last_check"] = {
|
|
"command": command,
|
|
"phase": phase,
|
|
"at": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
|
}
|
|
|
|
mutate_json(path, record)
|