atlas-iac/services/hermes/scripts/cli_lane_files.py

#!/usr/bin/env python3
"""Durable path and filesystem primitives for Hermes CLI lane journals."""

from __future__ import annotations

import json
import os
import re
import stat
import uuid
from pathlib import Path
from typing import Any

from cli_lane_config import STATE_ROOT, TerminalIdentity, canonical_run_id


def atomic_json(path: Path, value: dict[str, Any], mode: int = 0o600) -> None:
    """Durably replace a small state document without following temp symlinks."""
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    os.chmod(path.parent, 0o700, follow_symlinks=False)
    temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    flags |= getattr(os, "O_NOFOLLOW", 0)
    descriptor = None
    try:
        descriptor = os.open(temporary, flags, mode)
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            descriptor = None
            json.dump(value, stream, indent=2, sort_keys=True)
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, mode, follow_symlinks=False)
        os.replace(temporary, path)
        _fsync_directory(path.parent)
    finally:
        if descriptor is not None:
            os.close(descriptor)
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass

def _fsync_directory(directory: Path) -> None:
    """Persist directory-entry changes after an atomic rename or quarantine."""
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(directory, flags)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)

def load_json(path: Path) -> dict[str, Any]:
    try:
        flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
        descriptor = os.open(path, flags)
        with os.fdopen(descriptor, "r", encoding="utf-8") as stream:
            value = json.load(stream)
    except (OSError, UnicodeError, json.JSONDecodeError):
        return {}
    return value if isinstance(value, dict) else {}

def state_path(board: str, task_id: str) -> Path:
    safe_board = re.sub(r"[^a-zA-Z0-9_.-]+", "-", board)
    safe_task = re.sub(r"[^a-zA-Z0-9_.-]+", "-", task_id)
    return STATE_ROOT / safe_board / f"{safe_task}.json"

def _result_path(state_file: Path, run_id: Any, sequence: int) -> Path:
    """Return a unique provider-result path; completed turns are never reused."""
    safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown"))
    return state_file.with_name(
        f"{state_file.stem}.run-{safe_run}.provider-{sequence}.result.json"
    )

def _candidate_path(state_file: Path, run_id: Any, sequence: int) -> Path:
    """Return the durable path for one exact structured worker response."""
    safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown"))
    return state_file.with_name(
        f"{state_file.stem}.run-{safe_run}.candidate-{sequence}.json"
    )

def _terminal_path(
    state_file: Path,
    run_id: Any,
    state: str = "pending",
) -> Path:
    """Return the identity-bound replay path for an accepted terminal response."""
    if state not in {"pending", "committed"}:
        raise ValueError(f"invalid terminal journal state: {state}")
    canonical = canonical_run_id(run_id)
    if type(run_id) is not int or canonical is None:
        raise ValueError("terminal run ID must be a positive SQLite int64")
    return state_file.with_name(
        f"{state_file.stem}.run-{canonical}.terminal.{state}.json"
    )

_TERMINAL_NAME = re.compile(
    r"^(?P<task>[a-zA-Z0-9_.-]+)\.run-(?P<run>[0-9]+)\."
    r"terminal\.(?P<state>pending|committed)\.json$"
)
_TERMINAL_EVIDENCE_NAME = re.compile(
    r"^(?P<task>[a-zA-Z0-9_.-]+)\.run-(?P<run>[0-9]+)\."
    r"terminal\.(?P<state>prepared|conflict)-(?P<digest>[a-f0-9]{32})\.json$"
)
_RETIRE_NAME = re.compile(
    r"^\.retire\.(?P<path_digest>[a-f0-9]{16})\."
    r"(?P<source_digest>[a-f0-9]{16})\.(?P<sequence>[0-9]+)$"
)
_LEGACY_RETIRE_NAME = re.compile(r"^\.retire\.[a-f0-9]{32}\.[0-9]+$")
_SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$")


def _terminal_identity(path: Path) -> TerminalIdentity | None:
    """Parse replay authority lexically before opening a journal or board."""
    try:
        relative = path.relative_to(STATE_ROOT)
    except ValueError:
        return None
    if len(relative.parts) != 2:
        return None
    board, filename = relative.parts
    match = _TERMINAL_NAME.fullmatch(filename)
    run_id = canonical_run_id(match.group("run")) if match else None
    try:
        board_stat = path.parent.stat(follow_symlinks=False)
    except OSError:
        return None
    if (
        not match
        or run_id is None
        or not stat.S_ISDIR(board_stat.st_mode)
        or not _SAFE_BOARD.fullmatch(board)
        or board in {".", ".."}
        or match.group("task") in {".", ".."}
    ):
        return None
    return TerminalIdentity(
        board=board,
        task_id=match.group("task"),
        run_id=run_id,
        state=match.group("state"),
    )

def _terminal_evidence_identity(path: Path) -> TerminalIdentity | None:
    """Parse immutable prepared/conflict evidence authority from its path."""
    try:
        relative = path.relative_to(STATE_ROOT)
    except ValueError:
        return None
    if len(relative.parts) != 2:
        return None
    board, filename = relative.parts
    match = _TERMINAL_EVIDENCE_NAME.fullmatch(filename)
    run_id = canonical_run_id(match.group("run")) if match else None
    try:
        board_stat = path.parent.stat(follow_symlinks=False)
    except OSError:
        return None
    if (
        not match
        or run_id is None
        or not stat.S_ISDIR(board_stat.st_mode)
        or not _SAFE_BOARD.fullmatch(board)
        or board in {".", ".."}
        or match.group("task") in {".", ".."}
    ):
        return None
    return TerminalIdentity(
        board=board,
        task_id=match.group("task"),
        run_id=run_id,
        state=match.group("state"),
    )