gate_notify.py

#!/usr/bin/env python3
"""
gate_notify — shared voice throttle for the goal-contract Stop gates.  [2026-07-04]

Both disposition-gate.py and goal-gate.py are Stop hooks and both page the operator by
voice on a yield. When one session trips BOTH yields on the same Stop, two
voice-attention.py subprocesses spawn at once and talk over each other (the operator heard this
on the Hermes session, 2026-07-04). speak_once() collapses that: the first gate to fire
for a given (session, time-bucket) speaks; the other stays silent.

Atomic across the two independent hook processes via O_CREAT|O_EXCL on a bucketed lock —
no TOCTOU. Locks live in the OS temp dir (auto-reaped, and never in the audited
goal-contract tree). Fails silent: a voice page is best-effort, never load-bearing.
"""
import os, time, subprocess, tempfile

HOME = os.path.expanduser("~")
VOICE = os.path.join(HOME, ".config", "claude-hooks", "voice-attention.py")
LOCKDIR = os.path.join(tempfile.gettempdir(), "gc-voicelocks")
WINDOW = 6  # seconds; one voice per session per ~6s window across all gates


def _won_bucket(session_id, now, window=WINDOW):
    """True if THIS caller claimed the (session, bucket) lock — pure-ish, testable.
    `now` injectable so tests don't depend on wall clock."""
    os.makedirs(LOCKDIR, exist_ok=True)
    bucket = int(now // window)
    lock = os.path.join(LOCKDIR, f"{session_id}-{bucket}")
    try:
        fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        os.close(fd)
        return True
    except FileExistsError:
        return False


def speak_once(session_id, msg, window=WINDOW):
    """Page the operator by voice — but only if no sibling gate already spoke for this
    session in the current window. Best-effort; never raises."""
    try:
        if not _won_bucket(session_id or "unknown", time.time(), window):
            return  # a sibling gate already spoke this bucket
        if os.path.exists(VOICE):
            subprocess.Popen(["/usr/bin/python3", VOICE, "--say", msg],
                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except Exception:
        pass


if __name__ == "__main__":  # tiny self-check
    import shutil
    d = os.path.join(tempfile.gettempdir(), "gc-voicelocks")
    shutil.rmtree(d, ignore_errors=True)
    t = 1000.0
    assert _won_bucket("s1", t) is True,  "first caller wins"
    assert _won_bucket("s1", t) is False, "second caller same bucket stays silent"
    assert _won_bucket("s2", t) is True,  "different session still speaks"
    assert _won_bucket("s1", t + 6) is True, "same session next window speaks again"
    shutil.rmtree(d, ignore_errors=True)
    print("gate_notify self-check: all green")