bench/manager/core/watch.py

"""Disk watcher: the directories are the source of truth, so poll and narrate.

Catches moves the HTTP API never saw — a file dragged by hand, an agent,
another tool, or a pull from origin/main — and attributes them via the
expectations registered in state and the arrivals registered by sync.

Attribution is also the trigger gate. **State syncs; reactions don't**: a
move a pull applied is somebody else's action reaching this replica, so it
renders and narrates and nothing else — the side effects (opening a PR)
belong to the board whose user made the move. Every future automation hung
off a stage transition asks the same question here.
"""

from __future__ import annotations

import time

import agents
import config
import github
import state
import sync


def _board_sig() -> dict[str, set[str]]:
    sig = {}
    for slug in config.STAGE_DIRS:
        directory = config.TASKS / slug
        sig[slug] = {p.name for p in directory.glob("*.md")} if directory.is_dir() else set()
    return sig


def _actor(filename: str, stage: str) -> tuple[str, bool, bool]:
    """Who did this, whether it happened somewhere else, and whether it has
    already been narrated.

    A move this board made is claimed from the expectations; one a pull
    brought carries its commit author's name and is *remote* — this board
    is only rendering it; a plain mv on this disk is nobody in particular,
    but it is still this board's own disk, so it acts.

    *Quiet* is the mover saying it has told this story already: a phase
    ending sweeps its whole list into done/ and reports one ending. The
    line is skipped; nothing else about the move is.
    """
    actor, quiet = state.claim_move(filename, stage)
    if actor == "disk":
        who = sync.arrived_actor(filename)
        if who:
            return who, True, False
    return actor, False, quiet


def narrate(prev: dict[str, set[str]], cur: dict[str, set[str]]) -> None:
    """Two board signatures → the events between them."""
    prev_loc = {f: s for s, files in prev.items() for f in files}
    cur_loc = {f: s for s, files in cur.items() for f in files}
    for f, stage in sorted(cur_loc.items()):
        if f in prev_loc and prev_loc[f] != stage:
            actor, remote, quiet = _actor(f, stage)
            if not quiet:
                state.record_board_event({
                    "kind": "move", "file": f, "from": prev_loc[f], "to": stage,
                    "actor": actor, "remote": remote,
                    "summary": f"{f} moved {prev_loc[f]} → {stage} ({actor})",
                })
            # a failed run is worn by the card in the stage it died in —
            # wherever the card goes next, it arrives without the alarm
            agents.forget_failure(f)
            if stage == "review" and not remote:
                # a card entering review with a work branch gets a PR — on
                # the actor's board only, or the team gets one PR attempt
                # per replica
                github.open_pr_async(f)
        elif f not in prev_loc:
            actor, remote, quiet = _actor(f, stage)
            if quiet:
                continue
            state.record_board_event({
                "kind": "new", "file": f, "to": stage, "actor": actor,
                "remote": remote,
                "summary": f"{f} appeared in {stage}/"
                           + (f" ({actor})" if actor != "disk" else ""),
            })


def watcher(interval: float | None = None) -> None:
    interval = config.WATCH_INTERVAL if interval is None else interval
    prev = _board_sig()
    while True:
        time.sleep(interval)
        try:
            cur = _board_sig()
        except OSError:
            continue
        if cur == prev:
            continue
        narrate(prev, cur)
        prev = cur
        state.broadcast({"type": "board"})