SniperGold_ML/engine/lock.py

"""Run lock management (spec 21, G-11).

Heartbeat refresh every 60 s; lock is stale after 300 s without refresh.
Default policy REFUSES to take over a lock, even a stale one; release of a
stale lock requires the explicit human ``--force-release-lock`` flag. The
Engine never kills or takes over a live process automatically.
"""

import json
import os
import time


class LockError(Exception):
    pass


LOCK_HEARTBEAT_SEC = 60
LOCK_STALE_SEC = 300


def _read(path):
    try:
        with open(path, "rb") as fh:
            return json.loads(fh.read().decode("utf-8"))
    except FileNotFoundError:
        return None
    except Exception as e:
        raise LockError("unreadable lock file: %s" % e)


def lock_age(info):
    return time.time() - info.get("heartbeat_epoch_s", 0)


def lock_status(path):
    """Read-only lock inspection. Returns dict or None."""
    if not os.path.exists(path):
        return None
    info = _read(path)
    if info is None:
        return None
    info = dict(info)
    info["_age_sec"] = lock_age(info)
    info["_stale"] = info["_age_sec"] > LOCK_STALE_SEC
    return info


class LockHandle:
    def __init__(self, path, run_id, pid, force_release=False):
        self.path = path
        self.run_id = run_id
        self.pid = pid
        self.force_release = force_release
        self.held = False
        self._last_touch = 0.0

    def acquire(self):
        existing = _read(self.path)
        now = time.time()
        if existing is not None:
            age = now - existing.get("heartbeat_epoch_s", 0)
            if age <= LOCK_STALE_SEC:
                raise LockError("lock held by live run (pid %s, run %s)"
                                % (existing.get("pid"), existing.get("run_id")))
            if not self.force_release:
                raise LockError(
                    "stale lock present (age %.0f s); refusing takeover; "
                    "explicit --force-release-lock is required (human gate G-11)"
                    % age)
        os.makedirs(os.path.dirname(os.path.abspath(self.path)), exist_ok=True)
        self._write(now)
        self.held = True

    def _write(self, now):
        info = {"pid": self.pid, "run_id": self.run_id,
                "heartbeat_epoch_s": now}
        tmp = self.path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            json.dump(info, fh, sort_keys=True, separators=(",", ":"))
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(tmp, self.path)

    def touch(self):
        if not self.held:
            return
        now = time.time()
        if now - self._last_touch < LOCK_HEARTBEAT_SEC / 2.0:
            return
        self._write(now)
        self._last_touch = now

    def release(self):
        if not self.held:
            return
        try:
            os.remove(self.path)
        except FileNotFoundError:
            pass
        self.held = False