unbroker

1"""Storage helpers (stdlib only): atomic JSON, append-only JSONL, strict perms.2 3Default backend is local-json. The optional google-sheets tracker is handled in4report.py by emitting rows for the `google-workspace` skill; this module stays5dependency-free so the hermetic tests never touch the network.6"""7from __future__ import annotations8 9import contextlib10import json11import os12import time13from pathlib import Path14from typing import Any15 16import crypto17import paths18 19 20@contextlib.contextmanager21def locked(target: Path, timeout: float = 10.0, stale: float = 30.0):22    """Portable advisory lock via an O_EXCL lockfile next to `target`.23 24    Serializes read-modify-write on shared JSON (the ledger) across concurrent25    processes - a cron re-scan overlapping a manual run, or multiple tenants -26    so one writer can't clobber another's update. A lock older than `stale`27    seconds is treated as abandoned (crashed writer) and broken, so a dead28    process can never deadlock the queue. Works on macOS/Linux/Windows (O_EXCL).29    """30    ensure_dir(target.parent)31    lock = target.with_name(target.name + ".lock")32    deadline = time.monotonic() + timeout33    while True:34        try:35            fd = os.open(str(lock), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)36            try:37                os.write(fd, str(os.getpid()).encode())38            finally:39                os.close(fd)40            break41        except FileExistsError:42            try:43                if time.time() - lock.stat().st_mtime > stale:44                    lock.unlink(missing_ok=True)45                    continue46            except OSError:47                pass48            if time.monotonic() >= deadline:49                raise TimeoutError(f"could not acquire lock {lock} within {timeout}s")50            time.sleep(0.05)51    try:52        yield53    finally:54        with contextlib.suppress(OSError):55            lock.unlink(missing_ok=True)56 57 58def _secure(path: Path, mode: int) -> None:59    try:60        os.chmod(path, mode)61    except OSError:62        pass  # non-POSIX / unsupported FS; HERMES_HOME directory perms still apply63 64 65def ensure_dir(path: Path) -> Path:66    path.mkdir(parents=True, exist_ok=True)67    _secure(path, 0o700)68    return path69 70 71def _is_sensitive(path: Path) -> bool:72    """Per-subject docs (dossier, ledger) are sensitive; config/cache are not."""73    try:74        Path(path).resolve().relative_to(paths.subjects_dir().resolve())75        return True76    except (ValueError, OSError):77        return False78 79 80def _age_path(path: Path) -> Path:81    return path.with_name(path.name + ".age")82 83 84def _atomic_write(path: Path, data: bytes) -> Path:85    tmp = path.with_name(path.name + ".tmp")86    tmp.write_bytes(data)87    _secure(tmp, 0o600)88    os.replace(tmp, path)89    _secure(path, 0o600)90    return path91 92 93def write_json(path: Path, obj: Any) -> Path:94    ensure_dir(path.parent)95    data = (json.dumps(obj, indent=2, ensure_ascii=False) + "\n").encode("utf-8")96    if _is_sensitive(path) and crypto.encryption_setting() == "age":97        if not crypto.age_available():98            raise RuntimeError(99                "encryption=age is configured but `age` is not available; "100                "refusing to write PII as plaintext. Install age or run `setup --encryption none`."101            )102        target = _atomic_write(_age_path(path), crypto.encrypt(data))103        if path.exists():104            path.unlink()  # migrate plaintext -> ciphertext105        return target106    target = _atomic_write(path, data)107    ap = _age_path(path)108    if ap.exists():109        ap.unlink()  # encryption turned off -> drop stale ciphertext110    return target111 112 113def read_json(path: Path, default: Any = None) -> Any:114    ap = _age_path(path)115    if ap.exists():116        return json.loads(crypto.decrypt(ap.read_bytes()).decode("utf-8"))117    if path.exists():118        return json.loads(path.read_text(encoding="utf-8"))119    return default120 121 122def append_jsonl(path: Path, record: dict) -> Path:123    ensure_dir(path.parent)124    with path.open("a", encoding="utf-8") as fh:125        fh.write(json.dumps(record, ensure_ascii=False) + "\n")126    _secure(path, 0o600)127    return path128 129 130def read_jsonl(path: Path) -> list[dict]:131    if not path.exists():132        return []133    out: list[dict] = []134    for line in path.read_text(encoding="utf-8").splitlines():135        line = line.strip()136        if line:137            out.append(json.loads(line))138    return out139