rune/scripts/completion_file.py

"""completion_file.py — REAL_ASYNC_BRIDGE: durable completion-file delivery
with inotify wake. Replaces FIFO+lease for process-wait delivery.

DESIGN:
  Producer (bridge/daemon/notify):
    1. Validates operation identity
    2. Atomically writes a completion file:
         resumes/<operation_id>.done
       containing the event/verdict JSON (write to tmp, rename for atomicity)

  Waiter (ci_event_wait / any continuation consumer):
    1. Checks if the completion file already exists (event-before-waiter)
    2. If not: registers inotify watch on the resumes/ directory
    3. Re-checks after watch registration (closes the registration race)
    4. Blocks in kernel on inotify — zero CPU, zero timer
    5. On file-created event: reads the file, verifies authority, returns

  The event is a WAKE SIGNAL. Authority still decides terminal truth.
  The completion file is durable — survives daemon and waiter restarts.

WHY NOT FIFO:
  FIFO is one-shot (token consumed), requires lease management, and
  fails under session-busy guards. A completion file is atomic, durable,
  and works across restarts. inotify wakes the process with zero CPU
  cost and zero latency between file creation and wake.

WHY NOT POLLING:
  Polling introduces an arbitrary interval. inotify is a kernel callback
  that fires exactly when the filesystem changes — the latency tracks
  the real completion event, not a configured interval.
"""
from __future__ import annotations

import json
import os
import select
import struct
import time
from pathlib import Path

def _wr_bridge() -> Path:
    # WR_ROOT = daemon/infra convention; RUNE_WR_ROOT = repo/scripts
    # convention. Both honored (WR_ROOT wins) so the daemon, the waiter,
    # and any isolated test root always resolve the SAME directory.
    root = (os.environ.get("WR_ROOT")
            or os.environ.get("RUNE_WR_ROOT")
            or "/mnt/workdrive/wr-bridge")
    return Path(root)

def _resumes_dir() -> Path:
    return _wr_bridge() / "resumes"

# inotify constants (Linux)
IN_CREATE = 0x100
IN_CLOSE_WRITE = 0x8
IN_MOVED_TO = 0x80


def completion_path(operation_id: str) -> Path:
    return _resumes_dir() / f"{operation_id}.done"


def write_completion(operation_id: str, event: dict) -> Path:
    """Atomically write a durable completion file. The rename(2) ensures
    the waiter never sees a partial file."""
    _resumes_dir().mkdir(parents=True, exist_ok=True)
    target = completion_path(operation_id)
    tmp = target.parent / f".{target.name}.tmp.{os.getpid()}"
    tmp.write_text(json.dumps(event, sort_keys=True) + "\n")
    os.rename(str(tmp), str(target))  # atomic on same filesystem
    os.chmod(str(target), 0o666)  # cross-user bridge
    return target


def read_completion(operation_id: str) -> dict | None:
    """Read a durable completion file. Returns None if absent."""
    p = completion_path(operation_id)
    if not p.is_file():
        return None
    try:
        return json.loads(p.read_text())
    except (OSError, ValueError):
        return None


def completion_exists(operation_id: str) -> bool:
    return completion_path(operation_id).is_file()


def _inotify_init_and_add_watch(directory: str) -> int:
    """Create an inotify instance and add a watch on the directory.
    Returns the inotify fd."""
    libc = __import__("ctypes").CDLL(None, use_errno=True)
    fd = libc.inotify_init()
    if fd < 0:
        raise OSError("inotify_init failed")
    # IN_CREATE | IN_MOVED_TO: catch both direct creation and atomic rename
    mask = IN_CREATE | IN_MOVED_TO
    wd = libc.inotify_add_watch(fd, directory.encode(), mask)
    if wd < 0:
        os.close(fd)
        raise OSError(f"inotify_add_watch failed for {directory}")
    return fd


def _inotify_read(fd: int, timeout_secs: float | None = None) -> list[str]:
    """Read pending inotify events. Returns list of filenames that were
    created/moved-to. Blocks until at least one event or timeout."""
    poll = select.poll()
    poll.register(fd, select.POLLIN)
    events = []
    buf = os.read(fd, 4096)
    pos = 0
    while pos < len(buf):
        # struct: wd(i32), mask(u32), cookie(u32), name_len(u32)
        wd, mask, cookie, name_len = struct.unpack_from("iIII", buf, pos)
        pos += 16
        name = buf[pos:pos + name_len].rstrip(b"\x00").decode("utf-8", "replace")
        pos += name_len
        if mask & (IN_CREATE | IN_MOVED_TO):
            events.append(name)
    return events


def wait_for_completion(operation_id: str, *,
                        timeout_secs: float | None = None,
                        poll_callback=None) -> dict | None:
    """Block until a completion file exists for this operation.

    Uses inotify for zero-latency, zero-CPU wake. Falls back to
    polling_callback (if provided) when inotify is unavailable.

    The completion file is a WAKE SIGNAL. The caller must still verify
    canonical authority after reading it.

    Returns the event dict from the completion file, or None on timeout.
    """
    target = completion_path(operation_id)

    # 1. check-before-watch (event may have arrived before we started)
    existing = read_completion(operation_id)
    if existing is not None:
        return existing

    _resumes_dir().mkdir(parents=True, exist_ok=True)

    try:
        fd = _inotify_init_and_add_watch(str(_resumes_dir()))
    except OSError:
        # inotify unavailable: fall back to polling callback
        if poll_callback:
            return poll_callback(operation_id, timeout_secs)
        return None

    try:
        # 2. re-check after watch registration (closes the race between
        #    the initial check and the watch being active)
        existing = read_completion(operation_id)
        if existing is not None:
            return existing

        # 3. block in kernel on inotify — zero CPU, exact latency
        import select as _select
        poll = _select.poll()
        poll.register(fd, _select.POLLIN)
        deadline = (time.monotonic() + timeout_secs) if timeout_secs else None

        while True:
            remaining = None
            if deadline is not None:
                remaining = max(0.1, deadline - time.monotonic())
                if remaining <= 0.1:
                    return None  # timeout
            events = poll.poll(remaining * 1000 if remaining else None)
            if not events:
                continue  # poll timeout, re-check
            names = _inotify_read(fd)
            if target.name in names:
                # 4. file appeared — read it
                existing = read_completion(operation_id)
                if existing is not None:
                    return existing
                # partial/corrupt: keep watching
    finally:
        os.close(fd)
    return None


def clear_completion(operation_id: str) -> None:
    """Remove a consumed completion file (idempotent cleanup)."""
    p = completion_path(operation_id)
    try:
        p.unlink()
    except OSError:
        pass