fix(kanban): serialize shared board writes

hermes_cli/kanban_db.py

@@ -1347,6 +1347,7 @@ class Event:
 # lock (the in-process _INIT_LOCK + idempotent init remain the backstop).
 _INIT_LOCK_TIMEOUT_SECONDS = 10.0
 _INIT_LOCK_POLL_SECONDS = 0.05
+_WRITE_LOCK_LOCAL = threading.local()
 
 
 def _resolve_busy_timeout_ms() -> int:
@@ -1539,11 +1540,18 @@ def _dispatch_tick_lock(db_path: Path):
         acquired = True
         handle = None
     try:
+        if acquired:
+            held = getattr(_WRITE_LOCK_LOCAL, "held", set())
+            held.add(str(db_path.resolve()))
+            _WRITE_LOCK_LOCAL.held = held
         yield acquired
     finally:
         if handle is not None:
             try:
                 if acquired:
+                    held = getattr(_WRITE_LOCK_LOCAL, "held", set())
+                    held.discard(str(db_path.resolve()))
+                    _WRITE_LOCK_LOCAL.held = held
                     if _IS_WINDOWS:
                         import msvcrt
 
@@ -2736,6 +2744,38 @@ def _execute_boundary_with_retry(conn: sqlite3.Connection, sql: str) -> None:
             time.sleep(random.uniform(_BUSY_RETRY_MIN_S, _BUSY_RETRY_MAX_S))
 
 
+@contextlib.contextmanager
+def _common_write_lock(conn: sqlite3.Connection):
+    """Serialize every board write with the dispatcher's cross-process lock."""
+    db_path = Path(conn.execute("PRAGMA database_list").fetchone()[2]).resolve()
+    if str(db_path) in getattr(_WRITE_LOCK_LOCAL, "held", set()):
+        yield
+        return
+    lock_path = db_path.with_name(db_path.name + ".dispatch.lock")
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    handle = lock_path.open("a+b")
+    try:
+        if _IS_WINDOWS:
+            import msvcrt
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
+        else:
+            import fcntl
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+        yield
+    finally:
+        try:
+            if _IS_WINDOWS:
+                import msvcrt
+                handle.seek(0)
+                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                import fcntl
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+        finally:
+            handle.close()
+
+
 @contextlib.contextmanager
 def write_txn(conn: sqlite3.Connection):
     """Context manager for an IMMEDIATE write transaction.
@@ -2749,32 +2789,26 @@ def write_txn(conn: sqlite3.Connection):
     shadow the original exception with a spurious rollback error.
     """
     _assert_not_delegated_child_mutation()
-    _execute_boundary_with_retry(conn, "BEGIN IMMEDIATE")
-    try:
-        yield conn
-    except Exception:
-        try:
-            conn.execute("ROLLBACK")
-        except sqlite3.OperationalError:
-            # SQLite has already auto-rolled-back the transaction (typical
-            # under EIO, lock contention, or corruption). Nothing to undo;
-            # do not let this secondary failure shadow the real one.
-            pass
-        raise
-    else:
+    with _common_write_lock(conn):
+        _execute_boundary_with_retry(conn, "BEGIN IMMEDIATE")
         try:
-            _execute_boundary_with_retry(conn, "COMMIT")
+            yield conn
         except Exception:
-            # COMMIT exhausted retries with the txn still open; roll back so the
-            # connection isn't poisoned for the next BEGIN IMMEDIATE.
             try:
                 conn.execute("ROLLBACK")
             except sqlite3.OperationalError:
                 pass
             raise
-        # Post-commit file-length check: header page_count must match actual file pages.
-        # A discrepancy means a torn-extend — raise now rather than silently corrupt.
-        _check_file_length_invariant(conn)
+        else:
+            try:
+                _execute_boundary_with_retry(conn, "COMMIT")
+            except Exception:
+                try:
+                    conn.execute("ROLLBACK")
+                except sqlite3.OperationalError:
+                    pass
+                raise
+            _check_file_length_invariant(conn)
 
 
 # ---------------------------------------------------------------------------

tests/hermes_cli/test_kanban_dispatch_lock.py

@@ -12,6 +12,8 @@
 
 from __future__ import annotations
 
+import multiprocessing
+import time
 from pathlib import Path
 
 import pytest
@@ -101,3 +103,91 @@ def test_reentrant_same_path_lock_is_exclusive(conn):
         assert held_a is True
         with kb._dispatch_tick_lock(db_path) as held_b:
             assert held_b is False, "same-board lock must be exclusive"
+
+
+def test_write_txn_shares_the_dispatch_tick_lock_in_process(conn):
+    """``write_txn`` must take the *same* lock file as ``_dispatch_tick_lock``.
+
+    Regression for the gap described in FASE2_escopo_correcao.md: dashboard/
+    CLI writes (which go through ``write_txn``) previously had no cross-
+    process serialization against the dispatcher tick. This does not
+    exercise real OS-level blocking (see the multiprocessing test below for
+    that); it pins the invariant that both code paths resolve to the exact
+    same lock file for a given board, which is what makes them mutually
+    exclusive across processes.
+    """
+    db_path = kb.kanban_db_path(board="default")
+    dispatch_lock_path = db_path.with_name(db_path.name + ".dispatch.lock")
+    resolved = Path(conn.execute("PRAGMA database_list").fetchone()[2]).resolve()
+    write_lock_path = resolved.with_name(resolved.name + ".dispatch.lock")
+    assert write_lock_path == dispatch_lock_path.resolve()
+
+    kb.create_task(conn, title="t", assignee="w")
+    with kb.write_txn(conn):
+        conn.execute(
+            "INSERT INTO task_events (task_id, kind, payload, created_at) "
+            "VALUES (?, ?, ?, ?)",
+            ("lock-test", "test", None, 0),
+        )
+
+
+def _hold_dispatch_lock_then_signal(db_path_str, ready_evt, release_evt):
+    """Child process: acquire the board's dispatch lock and hold it."""
+    import fcntl
+
+    db_path = Path(db_path_str)
+    lock_path = db_path.with_name(db_path.name + ".dispatch.lock")
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    handle = lock_path.open("a+b")
+    fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+    ready_evt.set()
+    release_evt.wait(timeout=10)
+    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+    handle.close()
+
+
+def test_write_txn_blocks_behind_a_cross_process_dispatch_lock_holder(conn):
+    """A ``write_txn`` must block while another OS process holds the lock."""
+    db_path = kb.kanban_db_path(board="default")
+    kb.create_task(conn, title="t", assignee="w")
+
+    ready_evt = multiprocessing.Event()
+    release_evt = multiprocessing.Event()
+    proc = multiprocessing.Process(
+        target=_hold_dispatch_lock_then_signal,
+        args=(str(db_path), ready_evt, release_evt),
+    )
+    proc.start()
+    try:
+        assert ready_evt.wait(timeout=5), "child never acquired the dispatch lock"
+        started = time.monotonic()
+
+        import threading
+
+        def _release_after_delay():
+            time.sleep(0.3)
+            release_evt.set()
+
+        releaser = threading.Thread(target=_release_after_delay)
+        releaser.start()
+
+        with kb.write_txn(conn):
+            conn.execute(
+                "UPDATE tasks SET priority = 5 WHERE id = "
+                "(SELECT id FROM tasks LIMIT 1)"
+            )
+        elapsed = time.monotonic() - started
+        releaser.join()
+        assert elapsed >= 0.25, (
+            "write_txn returned before the external dispatch-lock holder "
+            "released it — the cross-process lock is not being honoured"
+        )
+    finally:
+        release_evt.set()
+        proc.join(timeout=5)
+        if proc.is_alive():
+            proc.terminate()
+            proc.join(timeout=5)
+
+    row = conn.execute("PRAGMA integrity_check").fetchone()
+    assert row[0] == "ok"