istos
9d98e2c55a
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.
90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""Shared in-memory state, persistence of event logs, and the SSE fan-out.
|
|
|
|
All cross-thread registries live here, guarded by LOCK where they are
|
|
mutated from several threads. Modules communicate through this state rather
|
|
than importing each other's internals.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import queue
|
|
import threading
|
|
import time
|
|
|
|
import config
|
|
|
|
LOCK = threading.Lock()
|
|
CLIENTS: set[queue.Queue] = set() # one queue per open SSE connection
|
|
SESSIONS: dict[str, dict] = {} # session_id -> meta
|
|
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.
|
|
serve_port = config.PORT
|
|
|
|
# The last card archived through this board — the scope of the ⌘Z undo.
|
|
LAST_ARCHIVED: dict | None = None
|
|
|
|
|
|
def broadcast(payload: dict) -> None:
|
|
msg = json.dumps(payload)
|
|
with LOCK:
|
|
clients = list(CLIENTS)
|
|
for q in clients:
|
|
try:
|
|
q.put_nowait(msg)
|
|
except queue.Full:
|
|
pass
|
|
|
|
|
|
def persist(name: str, record: dict) -> None:
|
|
try:
|
|
config.SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
|
|
with (config.SESSIONS_DIR / name).open("a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(record) + "\n")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def record_board_event(event: dict) -> None:
|
|
event["ts"] = time.time()
|
|
with LOCK:
|
|
BOARD_EVENTS.append(event)
|
|
del BOARD_EVENTS[:-config.BOARD_EVENTS_CAP]
|
|
persist("board.jsonl", event)
|
|
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:
|
|
EXPECTED_MOVES[(filename, target)] = (actor, time.time())
|
|
|
|
|
|
def claim_expected(filename: str, target: str) -> str:
|
|
with LOCK:
|
|
actor_ts = EXPECTED_MOVES.pop((filename, target), None)
|
|
# forget stale expectations while we're here
|
|
cutoff = time.time() - 30
|
|
for key in [k for k, (_, ts) in EXPECTED_MOVES.items() if ts < cutoff]:
|
|
EXPECTED_MOVES.pop(key, None)
|
|
return actor_ts[0] if actor_ts else "disk"
|