sync: origin/main as the shared board, behind BOARD_SYNC

Team mode's second half. A board-made move already commits itself (18);
now that commit publishes, every board pulls on a beat, and a card two
boards move at once resolves the way git resolves everything else — the
push race is the concurrency control.

- core/sync.py: push is event-driven (a new state.COMMIT_HOOKS registry
  fires it from taskfiles, so taskfiles stays left of everything that
  reacts to it); pull is a beat that fast-forwards, or replays this
  board's own commits on top when the two diverged. A replay that
  conflicts on a task file drops the local move — origin is the
  linearizer — and toasts who took the card.
- The piggyback guard stands in front of every push and every replay:
  each local-ahead commit on main must be `board: `-prefixed, so a
  human's unpushed work is never published as a side effect of a card
  moving. Uncommitted changes, a checkout off main and an unreachable
  origin all stall or degrade rather than risk anything, each narrated
  once instead of once per beat.
- watch.py names the commit author instead of "disk" for moves a pull
  brought, via the arrivals sync files for it; its narration moved out
  of the loop into narrate() so it can be tested directly.
- The board grows a sync chip that appears only when sync stops
  converging, and the SSE stream grows a toast type so the server can
  say something to the person, not just to the ticker.

Gate off (the default) means no fetch, no push, no thread, no change.

Verified with tests/test_boards_sync.py: two real clones of a real bare
upstream race each other through every case above.

manager/core/board.html

@@ -86,6 +86,9 @@
     background:var(--surface);border:1px solid var(--border);border-radius:99px;font-size:12.5px;
   }
   .livechip .mono{font-size:11.5px;color:var(--dim)}
+  /* an author display beats the UA's [hidden] — say it here or the sync
+     chip is never hidden */
+  .livechip[hidden]{display:none}
   .dot{width:7px;height:7px;border-radius:99px;background:var(--idle);flex:none}
   .dot.live{background:var(--accent);animation:breathe 2.4s ease-in-out infinite}
 
@@ -480,6 +483,7 @@
     <button data-view="focus">Focus</button>
   </nav>
   <span class="spacer"></span>
+  <div class="livechip" id="syncchip" hidden style="cursor:default"></div>
   <div class="livechip" id="livechip" title="open Sessions"></div>
   <button id="themebtn">Daylight</button>
   <button id="refresh">Refresh</button>
@@ -562,10 +566,10 @@ const STAGE_TINT = { backlog: 'var(--dim)', 'to-do': 'var(--muted)',
 const STAGE_NOTE = { 'to-do': 'next up', review: 'your move' };
 const GLYPHS = { session: '●', end: '○', idle: '…', edit: '✎', read: '◔', search: '⌕',
   command: '$', test: '▶', check: '☑', git: '⎇', plan: '≡', subagent: '⑂', web: '∿',
-  move: '⇢', new: '＋', agent: '⚑', report: '▣', other: '·' };
+  move: '⇢', new: '＋', agent: '⚑', report: '▣', sync: '⇅', other: '·' };
 const FILTERS = [
   ['all', 'All', null],
-  ['moves', 'Moves', new Set(['move', 'new', 'agent'])],
+  ['moves', 'Moves', new Set(['move', 'new', 'agent', 'sync'])],
   ['edits', 'Edits', new Set(['edit'])],
   ['reads', 'Reads', new Set(['read', 'search'])],
   ['tests', 'Tests', new Set(['test', 'check'])],
@@ -702,6 +706,10 @@ function connectStream() {
       scheduleRender();
     } else if (msg.type === 'board' || msg.type === 'agents') {
       loadState();
+    } else if (msg.type === 'toast') {
+      // the server needs to say something to the person, not just the
+      // ticker — losing a card to another board is the case that matters
+      toast(msg.message, !!msg.error);
     } else if (msg.type === 'board_event') {
       S.state?.boardEvents.push(msg.event);
       scheduleRender();
@@ -741,6 +749,23 @@ function renderChip() {
   }
 }
 
+/* Sync only shows itself when it has stopped converging: a stall that is
+   not visible is two halves of a team quietly drifting apart. Offline is
+   driftwood (degraded, self-healing); anything waiting on a human is
+   terracotta. */
+function renderSync() {
+  const s = S.state?.sync;
+  const el = $('#syncchip');
+  if (!s || !s.enabled || s.state === 'ok') { el.hidden = true; return; }
+  const alarm = s.state === 'stalled';
+  const detail = s.detail || '';
+  el.hidden = false;
+  el.title = detail;
+  el.innerHTML = `<span class="dot" style="background:var(--${alarm ? 'alarm' : 'idle'})"></span>` +
+    `<span style="color:var(--${alarm ? 'alarm' : 'muted'})">sync ${alarm ? 'stalled' : 'behind'}</span>` +
+    `<span class="mono">${esc(detail.split(' — ')[0].replace(/^sync[^:]*:\s*/, ''))}</span>`;
+}
+
 function setView(view) {
   S.view = view;
   document.querySelectorAll('#views button').forEach(b => b.classList.toggle('on', b.dataset.view === view));
@@ -762,6 +787,7 @@ function render() {
   if (!S.state) return;
   renderTitle();
   renderChip();
+  renderSync();
   if (S.view === 'board') renderBoard();
   else if (S.view === 'flight') renderFlight();
   else renderFocus();

manager/core/board.py

@@ -12,6 +12,7 @@ task files, but the tasks work as a plain folder kanban without it. See
     state.py      shared registries, event persistence, SSE fan-out
     taskfiles.py  reading/moving task files (the only code touching tasks/)
     events.py     hook payloads → displayable events, session registry
+    sync.py       origin/main as the shared board: push on move, pull on a beat
     agents.py     headless work/review agents: launch, reap, stop, diff
     watch.py      2s disk poller narrating moves made outside the API
     httpd.py      HTTP routes, SSE stream, the page itself
@@ -34,6 +35,7 @@ import events
 import github
 import httpd
 import state
+import sync
 import watch
 
 
@@ -64,6 +66,11 @@ def main() -> None:
     threading.Thread(target=watch.watcher, daemon=True).start()
     threading.Thread(target=github.poller, daemon=True).start()
     threading.Thread(target=github.reconcile, daemon=True).start()
+    if config.SYNC:
+        # Team mode: board commits publish themselves and a beat pulls what
+        # the other boards published. Off, neither thread nor hook exists.
+        sync.install()
+        threading.Thread(target=sync.beat, daemon=True).start()
     drive.adopt()
 
     print(f"Task board for {config.TASKS}\n  {url}\n  Ctrl-C to stop")

manager/core/config.py

@@ -134,12 +134,19 @@ PR_POLL_INTERVAL = float(setting("BOARD_PR_POLL_INTERVAL", "60"))
 # network weather; this bounds the whole delay.
 FETCH_TIMEOUT = float(setting("BOARD_FETCH_TIMEOUT", "10"))
 
+# Team mode's second half: origin/main is the shared truth and every board
+# a converging replica — board commits push as they are made, a beat pulls
+# what other boards published. Off by default; on, it implies COMMIT_MOVES
+# below, because a move that never commits has nothing to publish.
+SYNC = flag("BOARD_SYNC")
+SYNC_INTERVAL = float(setting("BOARD_SYNC_INTERVAL", "30"))
+
 # Team mode's first half: a board-made move claims the card (writing
 # **Assignee:** from git's own user.name) and commits itself, so ownership
 # and stage travel with the file to every clone. Off by default — a
 # single-player board moves cards exactly as it always did, and committing
 # tasks/ stays a hand job.
-COMMIT_MOVES = flag("BOARD_COMMIT_MOVES")
+COMMIT_MOVES = flag("BOARD_COMMIT_MOVES") or SYNC
 
 WATCH_INTERVAL = float(setting("BOARD_WATCH_INTERVAL", "2"))
 EVENTS_CAP = int(setting("BOARD_EVENTS_CAP", "800"))

manager/core/httpd.py

@@ -18,6 +18,7 @@ import drive
 import events
 import github
 import state
+import sync
 import taskfiles
 
 
@@ -40,6 +41,7 @@ def state_payload() -> dict:
         "commandRuns": commands.public(),
         "checks": config.checks(),
         "archivedCount": taskfiles.archived_count(),
+        "sync": sync.status(),
         "boardEvents": board_events,
         "now": time.time(),
     }

manager/core/state.py

@@ -21,6 +21,7 @@ EVENTS: dict[str, list[dict]] = {}         # session_id -> slim events
 BOARD_EVENTS: list[dict] = []              # moves + agent lifecycle
 AGENTS: dict[str, dict] = {}               # agent_id -> launch record
 EXPECTED_MOVES: dict[tuple[str, str], tuple[str, float]] = {}  # (file, to) -> (actor, ts)
+COMMIT_HOOKS: list = []                    # run after a board-made task commit
 
 # The port actually being served; board.py sets it from --port at startup so
 # launched agents know where to report events.
@@ -59,6 +60,19 @@ def record_board_event(event: dict) -> None:
     broadcast({"type": "board_event", "event": event})
 
 
+def task_committed(filename: str) -> None:
+    """A board-made move committed itself. Registered hooks turn that into
+    whatever else should follow — sync.py's push, when the gate is on. The
+    hook is a registry rather than an import so taskfiles stays to the left
+    of everything that reacts to it; a hook that raises must never break a
+    move that has already happened on disk."""
+    for hook in list(COMMIT_HOOKS):
+        try:
+            hook(filename)
+        except Exception:   # noqa: BLE001 — the move is done; nothing may undo it
+            pass
+
+
 def expect_move(filename: str, target: str, actor: str) -> None:
     """Tell the watcher who is about to move a file so it can attribute it."""
     with LOCK:

manager/core/sync.py

@@ -0,0 +1,448 @@
+"""Boards converge through origin/main: push what this board commits, pull
+what the other boards published.
+
+Gated on `BOARD_SYNC` (which implies `BOARD_COMMIT_MOVES` — a move that
+never commits has nothing to publish). Off, nothing here runs: no fetch,
+no push, no thread.
+
+The shape of it:
+
+- **push** is event-driven. `taskfiles` fires `state.task_committed` after
+  a board-made move commits; the hook installed here publishes it. A
+  rejected push means another board got there first, so the whole converge
+  runs and pushes again.
+- **pull** is a beat: fetch, then integrate. Purely behind → fast-forward.
+  Diverged → the board's own commits are rebased on top, never merged
+  past; a rebase that conflicts on a task file means the local move lost
+  the race, and it is dropped with a toast naming who took the card.
+- **the piggyback guard** stands in front of every push: each local-ahead
+  commit on main must be `board: `-prefixed. A human's unpushed work is
+  never published as a side effect of a card moving.
+- **offline** is not an error. The first unreachable fetch says so once,
+  the rest are silent, commits queue on local main and the next reachable
+  fetch catches up.
+
+Git is the lock server and main the linearizer — that is the whole
+concurrency control. Nothing here reacts to synced state beyond narrating
+it: replicas render, they do not act.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+import config
+import state
+from taskfiles import NUMBER_RE
+
+REMOTE = "origin"                     # one remote, one branch — by design
+BRANCH = "main"
+UPSTREAM = f"{REMOTE}/{BRANCH}"
+BOARD_COMMIT = "board: "              # the prefix taskfiles messages its own commits with
+PUSH_TIMEOUT = 120
+REBASE_TIMEOUT = 120
+ARRIVED_TTL = 60.0                    # the watcher polls every 2s; this is generous
+
+_LOCK = threading.Lock()              # one git operation on the checkout at a time
+_ARRIVED_LOCK = threading.Lock()
+ARRIVED: dict[str, tuple[str, float]] = {}   # filename -> (author, ts) from the last pull
+_NOTES: dict[str, tuple[str, str]] = {}      # key -> (summary, level) already narrated
+
+
+def _git(*args: str, timeout: float = 30) -> subprocess.CompletedProcess:
+    """Never raises: a timeout or a missing binary is just a failed run."""
+    try:
+        return subprocess.run(["git", "-C", str(config.REPO), *args],
+                              capture_output=True, text=True, timeout=timeout)
+    except subprocess.TimeoutExpired:
+        return subprocess.CompletedProcess(args, 128, "", "timed out")
+    except OSError as exc:
+        return subprocess.CompletedProcess(args, 128, "", str(exc))
+
+
+# ── narration ──────────────────────────────────────────────────────────
+# Every condition here repeats on every beat, so each one is narrated once
+# and then held: the ticker says it, the header chip keeps saying it.
+
+
+def status() -> dict:
+    """What the header shows: ok while converging, otherwise the reason."""
+    if not config.SYNC:
+        return {"enabled": False, "state": "off", "detail": ""}
+    for level in ("offline", "stalled"):
+        for summary, note_level in _NOTES.values():
+            if note_level == level:
+                return {"enabled": True, "state": level, "detail": summary}
+    return {"enabled": True, "state": "ok", "detail": ""}
+
+
+def _note(key: str, summary: str, level: str = "stalled") -> None:
+    if _NOTES.get(key) == (summary, level):
+        return          # same condition as last time: said once is enough
+    _NOTES[key] = (summary, level)
+    state.record_board_event({"kind": "sync", "actor": "sync", "summary": summary})
+    state.broadcast({"type": "board"})
+
+
+def _clear(key: str, recovery: str = "") -> None:
+    if _NOTES.pop(key, None) is None:
+        return
+    if recovery:
+        state.record_board_event({"kind": "sync", "actor": "sync", "summary": recovery})
+    state.broadcast({"type": "board"})
+
+
+# ── the checkout ───────────────────────────────────────────────────────
+
+
+def _origin_present() -> bool:
+    return REMOTE in _git("remote").stdout.split()
+
+
+def _head() -> str:
+    return _git("rev-parse", "HEAD").stdout.strip()
+
+
+def _on_main() -> bool:
+    return _git("rev-parse", "--abbrev-ref", "HEAD").stdout.strip() == BRANCH
+
+
+def _clean() -> bool:
+    """Tracked files only: an untracked scratch file is nobody's business,
+    but a modified one is what a fast-forward would run over."""
+    return not _git("status", "--porcelain", "--untracked-files=no").stdout.strip()
+
+
+def _count(rng: str) -> int:
+    out = _git("rev-list", "--count", rng).stdout.strip()
+    return int(out) if out.isdigit() else 0
+
+
+def _tasks_prefix() -> str:
+    try:
+        return config.TASKS.resolve().relative_to(config.REPO.resolve()).as_posix() + "/"
+    except ValueError:
+        return "tasks/"
+
+
+def _fetch() -> bool:
+    result = _git("fetch", REMOTE, BRANCH, timeout=config.FETCH_TIMEOUT)
+    if result.returncode != 0:
+        if "couldn't find remote ref" in (result.stderr or "").lower():
+            _note("no-branch", f"sync stalled: {REMOTE} has no {BRANCH} branch — "
+                               f"sync rides {UPSTREAM} and nothing else")
+            return False
+        _note("offline",
+              f"sync is behind: {REMOTE} is unreachable — this board keeps "
+              f"working locally and catches up when it returns", "offline")
+        return False
+    _clear("no-branch")
+    _clear("offline", f"sync caught up: {REMOTE} is reachable again")
+    return True
+
+
+# ── publishing ─────────────────────────────────────────────────────────
+
+
+def _ahead() -> list[str]:
+    """`<short sha> <subject>` for every commit local main has and
+    origin/main does not — newest first."""
+    out = _git("log", "--format=%h %s", f"{UPSTREAM}..{BRANCH}").stdout
+    return [line for line in out.splitlines() if line.strip()]
+
+
+def _stray(commits: list[str]) -> str:
+    """The piggyback hazard: pushing publishes *every* local-ahead commit,
+    so one that the board did not make is a human's private work and stops
+    the push. Oldest first — that is the one to deal with."""
+    for line in reversed(commits):
+        subject = line.split(" ", 1)[1] if " " in line else ""
+        if not subject.startswith(BOARD_COMMIT):
+            return line
+    return ""
+
+
+def _publish() -> str:
+    """Push local main if — and only if — everything on it is the board's.
+
+    ok | nothing | stray | not-on-main | retry | offline | stalled
+    """
+    if not _on_main():
+        branch = _git("rev-parse", "--abbrev-ref", "HEAD").stdout.strip() or "a detached HEAD"
+        _note("branch", f"sync paused: this checkout is on '{branch}', not {BRANCH} — "
+                        f"board commits are not landing where sync publishes from")
+        return "not-on-main"
+    _clear("branch")
+    if _git("rev-parse", "--verify", "--quiet", UPSTREAM).returncode != 0:
+        return "retry"          # never fetched: converge first, then publish
+    commits = _ahead()
+    stray = _stray(commits)
+    if stray:
+        _note("stray", f"not pushing: {stray} is not a board commit — sync will not "
+                       f"publish it for you. Push main yourself, or move that commit "
+                       f"off main, and sync resumes")
+        return "stray"
+    _clear("stray")     # nothing stray left to refuse, however that happened
+    if not commits:
+        return "nothing"
+
+    result = _git("push", REMOTE, f"{BRANCH}:{BRANCH}", timeout=PUSH_TIMEOUT)
+    if result.returncode == 0:
+        _clear("push")
+        _clear("offline", f"sync caught up: {REMOTE} is reachable again")
+        state.record_board_event({
+            "kind": "sync", "actor": "sync",
+            "summary": f"pushed {len(commits)} board commit"
+                       f"{'s' if len(commits) > 1 else ''} to {UPSTREAM}"})
+        return "ok"
+    stderr = (result.stderr or result.stdout).strip()
+    if _rejected(stderr):
+        return "retry"
+    if _unreachable(stderr):
+        _note("offline",
+              f"sync is behind: {REMOTE} is unreachable — this board keeps "
+              f"working locally and catches up when it returns", "offline")
+        return "offline"
+    detail = stderr.splitlines()[-1][:140] if stderr else "git said nothing"
+    _note("push", f"sync could not push to {UPSTREAM}: {detail}")
+    return "stalled"
+
+
+def _rejected(stderr: str) -> bool:
+    text = stderr.lower()
+    return "non-fast-forward" in text or "fetch first" in text or "! [rejected]" in text
+
+
+def _unreachable(stderr: str) -> bool:
+    text = stderr.lower()
+    return any(mark in text for mark in (
+        "could not read from remote", "could not resolve", "unable to access",
+        "does not appear to be a git repository", "connection", "timed out",
+        "no such file or directory", "permission denied"))
+
+
+# ── integrating what arrived ───────────────────────────────────────────
+
+
+def _conflicted() -> list[str]:
+    out = _git("diff", "--name-only", "--diff-filter=U").stdout
+    return [line.strip() for line in out.splitlines() if line.strip()]
+
+
+def _is_task_file(path: str) -> bool:
+    return path.startswith(_tasks_prefix()) and path.endswith(".md")
+
+
+def _author_of(filename: str, rev: str) -> str:
+    """Who wrote the newest commit touching this card in `rev` — a range for
+    what a pull brought, a ref for what origin already holds."""
+    return _git("log", "-1", "--format=%an", rev, "--",
+                f"{_tasks_prefix()}*/{filename}").stdout.strip()
+
+
+def _number(filename: str) -> str:
+    match = NUMBER_RE.match(filename)
+    return match.group(1) if match else filename[:-3] if filename.endswith(".md") else filename
+
+
+def _lost(filename: str) -> None:
+    """The local move lost the race. Say who took the card — the file itself
+    reverts to origin's version when the rebase drops our commit."""
+    who = _author_of(filename, UPSTREAM) or "someone else"
+    message = f"{_number(filename)} claimed by {who} — your move was undone"
+    state.record_board_event({"kind": "sync", "actor": "sync", "file": filename,
+                              "summary": message})
+    state.broadcast({"type": "toast", "message": message, "error": True})
+    state.broadcast({"type": "board"})
+
+
+def _replay() -> str:
+    """Rebase this board's commits onto origin/main. Conflicts on a task
+    file are resolved by dropping our commit: origin is the linearizer, and
+    a card someone else moved first is theirs. Anything conflicting outside
+    tasks/ is a real collision — abort and wait for a human.
+
+    ok | dirty | stalled
+    """
+    if not _clean():
+        _note("dirty", "sync paused: main has uncommitted changes — commit or stash "
+                       "them and sync resumes (code work belongs in a worktree)")
+        return "dirty"
+    _clear("dirty")
+
+    result = _git("rebase", UPSTREAM, timeout=REBASE_TIMEOUT)
+    for _ in range(50):                    # bounded: one round per replayed commit
+        if result.returncode == 0:
+            _clear("replay")
+            return "ok"
+        conflicted = _conflicted()
+        if not conflicted or not all(_is_task_file(p) for p in conflicted):
+            _git("rebase", "--abort")
+            detail = ", ".join(conflicted[:3]) or (result.stderr or result.stdout).strip()[-140:]
+            _note("replay", f"sync stalled: replaying this board's commits onto {UPSTREAM} "
+                            f"collides outside tasks/ ({detail}) — a human has to settle it")
+            return "stalled"
+        for name in dict.fromkeys(Path(p).name for p in conflicted):
+            _lost(name)
+        result = _git("rebase", "--skip", timeout=REBASE_TIMEOUT)
+    _git("rebase", "--abort")
+    _note("replay", f"sync stalled: replaying onto {UPSTREAM} did not settle — "
+                    f"a human has to settle it")
+    return "stalled"
+
+
+def _integrate() -> str:
+    """Bring local main to origin/main without ever merging past a
+    divergence.
+
+    up-to-date | pulled | not-on-main | dirty | diverged | stalled
+    """
+    if _count(f"{BRANCH}..{UPSTREAM}") == 0:
+        return "up-to-date"
+    if not _on_main():
+        branch = _git("rev-parse", "--abbrev-ref", "HEAD").stdout.strip() or "a detached HEAD"
+        _note("branch", f"sync paused: this checkout is on '{branch}', not {BRANCH} — "
+                        f"switch back and the board catches up with {UPSTREAM}")
+        return "not-on-main"
+    _clear("branch")
+
+    # Diverged. The board's own bookkeeping can be replayed on top of what
+    # arrived — that is how a lost race resolves. A human's commit cannot,
+    # and the guard that refuses to push it refuses to rebase it too.
+    commits = _ahead()
+    stray = _stray(commits)
+    if stray:
+        _note("diverged", f"sync stalled: main and {UPSTREAM} have diverged and "
+                          f"{stray} is not a board commit — pull or rebase it by "
+                          f"hand, and this board starts converging again")
+        return "diverged"
+    _clear("diverged")
+    if commits:
+        outcome = _replay()
+        return "pulled" if outcome == "ok" else outcome
+
+    if not _clean():
+        _note("dirty", "sync paused: main has uncommitted changes — commit or stash "
+                       "them and sync resumes (code work belongs in a worktree)")
+        return "dirty"
+    _clear("dirty")
+    result = _git("merge", "--ff-only", UPSTREAM, timeout=REBASE_TIMEOUT)
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip().splitlines()
+        _note("merge", f"sync stalled: fast-forwarding to {UPSTREAM} failed "
+                       f"({detail[-1][:140] if detail else 'no detail'})")
+        return "stalled"
+    _clear("merge")
+    return "pulled"
+
+
+def _record_arrivals(before: str) -> None:
+    """Attribute what the pull brought: each task file it touched is filed
+    under the name of whoever committed it, for the watcher to use instead
+    of "disk" when the move surfaces on the next poll."""
+    head = _head()
+    if not before or not head or head == before:
+        return
+    rng = f"{before}..{head}"
+    changed = _git("diff", "--name-only", rng, "--", _tasks_prefix()).stdout.splitlines()
+    names = sorted({Path(p).name for p in changed if p.strip().endswith(".md")})
+    if not names:
+        return
+    now = time.time()
+    authors = set()
+    with _ARRIVED_LOCK:
+        for name in names:
+            who = _author_of(name, rng)
+            if who:
+                ARRIVED[name] = (who, now)
+                authors.add(who)
+        for name in [n for n, (_, ts) in ARRIVED.items() if now - ts > ARRIVED_TTL]:
+            ARRIVED.pop(name, None)
+    count = _count(rng)
+    state.record_board_event({
+        "kind": "sync", "actor": "sync",
+        "summary": f"pulled {count} commit{'s' if count != 1 else ''} from {UPSTREAM}"
+                   + (f" ({', '.join(sorted(authors))})" if authors else "")})
+    state.broadcast({"type": "board"})
+
+
+def arrived_actor(filename: str) -> str:
+    """Who moved this card, if a pull just brought it. Consumed once — the
+    watcher asks exactly when it notices the move."""
+    with _ARRIVED_LOCK:
+        entry = ARRIVED.pop(filename, None)
+    if not entry:
+        return ""
+    who, ts = entry
+    return who if time.time() - ts <= ARRIVED_TTL else ""
+
+
+# ── the two entry points ───────────────────────────────────────────────
+
+
+def _converge() -> str:
+    """One full beat: fetch, integrate what arrived, publish what is ours."""
+    if not _origin_present():
+        return "no-origin"
+    if not _fetch():
+        return "offline"
+    before = _head()
+    outcome = _integrate()
+    _record_arrivals(before)
+    if outcome in ("up-to-date", "pulled"):
+        published = _publish()
+        if published in ("stray", "offline", "stalled", "not-on-main"):
+            return published
+    return outcome
+
+
+def push_now() -> str:
+    """A board commit just landed — publish it. The fast path skips the
+    fetch; a rejection means another board pushed first, and then the full
+    converge (fetch, replay, push) runs."""
+    if not config.SYNC:
+        return "off"
+    with _LOCK:
+        if not _origin_present():
+            return "no-origin"
+        outcome = _publish()
+        if outcome != "retry":
+            return outcome
+        return _converge()
+
+
+def pull_now() -> str:
+    """The beat. Also the offline catch-up: a fetch that works again is
+    followed by the push that could not happen while origin was gone."""
+    if not config.SYNC:
+        return "off"
+    with _LOCK:
+        return _converge()
+
+
+def on_commit(filename: str) -> None:
+    """The `state.task_committed` hook: publish off the caller's thread, so
+    a card move never waits on the network."""
+    if not config.SYNC:
+        return
+    threading.Thread(target=push_now, name="sync-push", daemon=True).start()
+
+
+def install() -> None:
+    """Wire the push hook. Called once at startup, only with the gate on."""
+    if on_commit not in state.COMMIT_HOOKS:
+        state.COMMIT_HOOKS.append(on_commit)
+
+
+def beat(interval: float | None = None) -> None:
+    interval = config.SYNC_INTERVAL if interval is None else interval
+    while True:
+        try:
+            pull_now()
+        except Exception as exc:              # noqa: BLE001 — the beat outlives a bad cycle
+            state.record_board_event({"kind": "sync", "actor": "sync",
+                                      "summary": f"sync cycle failed: {str(exc)[:140]}"})
+        time.sleep(interval)

manager/core/taskfiles.py

@@ -201,6 +201,7 @@ def _commit_move(filename: str, target: str, src: Path, dst: Path, who: str,
         if result.returncode == 0:
             result = _git("commit", "--no-verify", "-m", message, "--", *spec, timeout=60)
             if result.returncode == 0:
+                state.task_committed(filename)   # sync (when on) publishes it
                 return
         lines = (result.stderr or result.stdout).strip().splitlines()
         detail = lines[-1] if lines else f"git exited {result.returncode}"

manager/core/watch.py

@@ -1,7 +1,8 @@
 """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, or
-another tool — and attributes them via the expectations registered in state.
+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.
 """
 
 from __future__ import annotations
@@ -11,6 +12,7 @@ import time
 import config
 import github
 import state
+import sync
 
 
 def _board_sig() -> dict[str, set[str]]:
@@ -21,6 +23,40 @@ def _board_sig() -> dict[str, set[str]]:
     return sig
 
 
+def _actor(filename: str, stage: str) -> str:
+    """Who did this. A move this board made is claimed from the
+    expectations; one a pull brought carries its commit author's name; a
+    plain mv on this disk is nobody in particular."""
+    actor = state.claim_expected(filename, stage)
+    if actor == "disk":
+        return sync.arrived_actor(filename) or "disk"
+    return actor
+
+
+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 = _actor(f, stage)
+            state.record_board_event({
+                "kind": "move", "file": f, "from": prev_loc[f], "to": stage,
+                "actor": actor,
+                "summary": f"{f} moved {prev_loc[f]} → {stage} ({actor})",
+            })
+            if stage == "review":
+                # a card entering review with a work branch gets a PR
+                github.open_pr_async(f)
+        elif f not in prev_loc:
+            actor = _actor(f, stage)
+            state.record_board_event({
+                "kind": "new", "file": f, "to": stage, "actor": actor,
+                "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()
@@ -32,23 +68,6 @@ def watcher(interval: float | None = None) -> None:
             continue
         if cur == prev:
             continue
-        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 = state.claim_expected(f, stage)
-                state.record_board_event({
-                    "kind": "move", "file": f, "from": prev_loc[f], "to": stage,
-                    "actor": actor,
-                    "summary": f"{f} moved {prev_loc[f]} → {stage} ({actor})",
-                })
-                if stage == "review":
-                    # a card entering review with a work branch gets a PR
-                    github.open_pr_async(f)
-            elif f not in prev_loc:
-                state.record_board_event({
-                    "kind": "new", "file": f, "to": stage, "actor": "disk",
-                    "summary": f"{f} appeared in {stage}/",
-                })
+        narrate(prev, cur)
         prev = cur
         state.broadcast({"type": "board"})