fix(concurrency): fleet-safe decisions + crash windows (step 1, part 1)

docs/multi-agent.md

@@ -29,17 +29,31 @@ running Cursor".
 
 ## Concurrency
 
-vouch is single-writer per file. Two agents proposing simultaneously
-each get their own proposal id; no lock contention. Two agents
-approving simultaneously can race — the second one through sees
-`already_decided`.
-
-State.db (the FTS5 index) is updated under SQLite's normal locking.
-Reads do not block reads; writes briefly block other writes.
-
-Bottom line: don't worry about concurrent agents writing. *Do* worry
-about concurrent approvals if you have a script approving in bulk —
-serialise them.
+Five agents against one `.vouch/` is the configuration vouch is built
+for, so concurrent access is handled in the storage layer rather than
+left to you.
+
+**Proposing is lock-free.** Two agents proposing at the same time each
+get their own proposal id; there's nothing to contend over.
+
+**Durable writes are atomic.** Every artifact is written to a temp
+file, fsynced, then renamed into place, so a crash leaves either the
+old bytes or the complete new bytes — never a half-written file, and
+never an audit event attesting to yaml that power loss erased.
+
+**Approvals are serialised.** `approve`, `reject`, and expiry take a
+per-KB decision lock (`decisions.lock`) across the whole
+read-check-write, so two agents deciding the same proposal at once
+can't both proceed: one wins, the other finds it already decided. You
+don't need to serialise a bulk-approval script yourself — vouch does it.
+
+**The search index never blocks a write.** `state.db` (the FTS5 +
+embedding index) runs in WAL mode with a busy timeout, so readers and
+the single writer don't block each other and a contended writer waits
+briefly instead of failing with "database is locked". The index is
+derived and best-effort: if an index write fails during an approval,
+the approval still lands durably — rebuild the missing row with
+`vouch index`.
 
 ## Approval policies
 

src/vouch/audit.py

@@ -25,6 +25,7 @@
 
 AUDIT_FILENAME = "audit.log.jsonl"
 AUDIT_LOCKFILE = AUDIT_FILENAME + ".lock"
+DECISION_LOCKFILE = "decisions.lock"
 GENESIS_HASH = "0" * 64
 
 
@@ -37,16 +38,13 @@ def _audit_lockfile(kb_dir: Path) -> Path:
 
 
 @contextlib.contextmanager
-def _audit_lock(kb_dir: Path) -> Iterator[None]:
-    """Hold an exclusive cross-process lock for the duration of a log_event.
+def _flock(lockfile: Path) -> Iterator[None]:
+    """Hold an exclusive cross-process lock on ``lockfile`` for the block.
 
-    Serialises the read-then-append sequence in `log_event` so two concurrent
-    writers cannot both observe the same `prev_hash` and fork the chain.
-    Lock is held on a sibling `audit.log.jsonl.lock` file so the audit log
-    itself is never opened in a mode that could truncate it. Blocks until
-    acquired and is always released, including on exceptions.
+    Blocks until acquired and is always released, including on exceptions.
+    POSIX uses flock(2); Windows uses msvcrt byte-range locking on the same
+    descriptor.
     """
-    lockfile = _audit_lockfile(kb_dir)
     lockfile.parent.mkdir(parents=True, exist_ok=True)
     fd = os.open(lockfile, os.O_RDWR | os.O_CREAT, 0o644)
     try:
@@ -71,6 +69,34 @@ def _audit_lock(kb_dir: Path) -> Iterator[None]:
         os.close(fd)
 
 
+@contextlib.contextmanager
+def _audit_lock(kb_dir: Path) -> Iterator[None]:
+    """Hold an exclusive cross-process lock for the duration of a log_event.
+
+    Serialises the read-then-append sequence in `log_event` so two concurrent
+    writers cannot both observe the same `prev_hash` and fork the chain.
+    Lock is held on a sibling `audit.log.jsonl.lock` file so the audit log
+    itself is never opened in a mode that could truncate it.
+    """
+    with _flock(_audit_lockfile(kb_dir)):
+        yield
+
+
+@contextlib.contextmanager
+def decision_lock(kb_dir: Path) -> Iterator[None]:
+    """Serialise the review-gate decisions (approve/reject/expire) on a KB.
+
+    The read-check-write in approve()/reject()/expire_one() is not atomic, so
+    two processes — console, mcp, cli — can both read a proposal as pending and
+    both decide it, leaving a durable claim under a REJECTED record. Holding
+    this per-KB lock across a whole decision makes the second caller observe the
+    first's result. A distinct lockfile from the audit lock; log_event's
+    `_audit_lock` nests inside it in a consistent order, so there is no deadlock.
+    """
+    with _flock(kb_dir / DECISION_LOCKFILE):
+        yield
+
+
 def new_event_id() -> str:
     return uuid.uuid4().hex
 

src/vouch/index_db.py

@@ -99,6 +99,14 @@ def open_db(kb_dir: Path):
     path = _db_path(kb_dir)
     path.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(str(path))
+    # Fleet-safe defaults: five concurrent agents on one KB is the dogfood
+    # configuration. WAL lets readers proceed while a single writer commits
+    # (instead of a whole-file lock), and busy_timeout makes a contended writer
+    # wait for the lock rather than failing immediately with "database is
+    # locked". state.db is derived, so the -wal/-shm sidecars are gitignored
+    # (state.db-*). Set before any statement opens a transaction.
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA busy_timeout=5000")
     try:
         conn.executescript(SCHEMA)
         yield conn

src/vouch/proposals.py

@@ -6,8 +6,11 @@
 
 from __future__ import annotations
 
+import logging
+import sqlite3
 import time
 import uuid
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from typing import Any
@@ -20,6 +23,7 @@
     Claim,
     Entity,
     Page,
+    PageStatus,
     Proposal,
     ProposalKind,
     ProposalStatus,
@@ -28,6 +32,8 @@
 from .page_kinds import PageKindError, load_page_kind_registry, validate_page
 from .storage import ArtifactNotFoundError, KBStore
 
+_log = logging.getLogger("vouch.proposals")
+
 
 class ProposalError(RuntimeError):
     pass
@@ -497,6 +503,34 @@ def check_approvable(
     return _payload_block_reason(store, proposal)
 
 
+def _index_or_degrade(
+    store: KBStore,
+    kind: str,
+    artifact_id: str,
+    index_fn: Callable[[sqlite3.Connection], None],
+) -> None:
+    """Write an approved artifact's FTS row, degrading instead of raising.
+
+    state.db is derived, so a locked or broken index must not brick the
+    durable approve: put_<kind>() has already written the artifact, and the
+    caller still has to record the decision and log the approve event. Letting
+    the index error escape would abort between the artifact write and
+    move_proposal_to_decided(), leaving the file on disk with the proposal
+    still pending -- and the retry would hit _ensure_no_existing_artifact and
+    brick the approve for good. Rebuild the missing row later with
+    `vouch index`.
+    """
+    try:
+        with index_db.open_db(store.kb_dir) as conn:
+            index_fn(conn)
+    except sqlite3.Error as e:
+        _log.warning(
+            "%s %s: FTS5 index skipped on approve (%s); state.db is derived, "
+            "rebuild with `vouch index`",
+            kind, artifact_id, e,
+        )
+
+
 def approve(
     store: KBStore,
     proposal_id: str,
@@ -508,7 +542,23 @@ def approve(
 
     Raises ProposalError if the proposal is not pending or if
     approved_by matches proposed_by (forbidden_self_approval).
+
+    The whole read-check-write runs under the per-KB decision lock so a
+    concurrent reject or approve of the same proposal cannot both proceed.
     """
+    with audit.decision_lock(store.kb_dir):
+        return _approve_locked(
+            store, proposal_id, approved_by=approved_by, reason=reason
+        )
+
+
+def _approve_locked(
+    store: KBStore,
+    proposal_id: str,
+    *,
+    approved_by: str,
+    reason: str | None = None,
+) -> Claim | Page | Entity | Relation:
     proposal = store.get_proposal(proposal_id)
     block = _approval_block_reason(store, proposal, approved_by)
     if block:
@@ -532,14 +582,21 @@ def approve(
             **payload
         )
         store.put_claim(claim)
-        with index_db.open_db(store.kb_dir) as conn:
-            index_db.index_claim(
+        _index_or_degrade(
+            store, "claim", claim.id,
+            lambda conn: index_db.index_claim(
                 conn, id=claim.id, text=claim.text,
-                type=claim.type.value, status=claim.status.value, tags=claim.tags,
-            )
+                type=claim.type.value, status=claim.status.value,
+                tags=claim.tags,
+            ),
+        )
         result = claim
     elif proposal.kind == ProposalKind.PAGE:
         page = Page(**payload)
+        # Approval is what makes a page live. Page defaults to DRAFT and nothing
+        # promoted it, so reviewed pages read as draft forever; stamp ACTIVE at
+        # the gate.
+        page.status = PageStatus.ACTIVE
         # Re-validate the kind at the gate: config may have tightened (or a
         # kind been removed) between propose and approve. Built-in kinds pass
         # trivially, so this is a no-op for the common path.
@@ -562,11 +619,13 @@ def approve(
             store.update_page(page)
         except ArtifactNotFoundError:
             store.put_page(page)
-        with index_db.open_db(store.kb_dir) as conn:
-            index_db.index_page(
+        _index_or_degrade(
+            store, "page", page.id,
+            lambda conn: index_db.index_page(
                 conn, id=page.id, title=page.title, body=page.body,
                 type=page.type, tags=page.tags,
-            )
+            ),
+        )
         result = page
         # Lazy import: extractors.edges calls back into propose_relation,
         # so importing it at module scope would be circular.
@@ -576,11 +635,14 @@ def approve(
     elif proposal.kind == ProposalKind.ENTITY:
         entity = Entity(**payload)
         store.put_entity(entity)
-        with index_db.open_db(store.kb_dir) as conn:
-            index_db.index_entity(
-                conn, id=entity.id, name=entity.name, description=entity.description,
+        _index_or_degrade(
+            store, "entity", entity.id,
+            lambda conn: index_db.index_entity(
+                conn, id=entity.id, name=entity.name,
+                description=entity.description,
                 type=entity.type.value, aliases=entity.aliases,
-            )
+            ),
+        )
         result = entity
     elif proposal.kind == ProposalKind.DELETE:
         result = _approve_delete(store, proposal, approved_by=approved_by)
@@ -611,22 +673,25 @@ def reject(
 ) -> Proposal:
     if not reason.strip():
         raise ProposalError("rejection must include a reason (future agent context)")
-    proposal = store.get_proposal(proposal_id)
-    if proposal.status != ProposalStatus.PENDING:
-        raise ProposalError(
-            f"proposal {proposal_id} is {proposal.status.value}, not pending"
+    # Serialise with approve()/expire_one() so a concurrent approve of this same
+    # proposal cannot land a durable claim under this REJECTED record.
+    with audit.decision_lock(store.kb_dir):
+        proposal = store.get_proposal(proposal_id)
+        if proposal.status != ProposalStatus.PENDING:
+            raise ProposalError(
+                f"proposal {proposal_id} is {proposal.status.value}, not pending"
+            )
+        proposal.status = ProposalStatus.REJECTED
+        proposal.decided_at = datetime.now(UTC)
+        proposal.decided_by = rejected_by
+        proposal.decision_reason = reason
+        store.move_proposal_to_decided(proposal)
+        audit.log_event(
+            store.kb_dir, event=f"proposal.{proposal.kind.value}.reject",
+            actor=rejected_by, object_ids=[proposal.id],
+            data={"reason": reason},
         )
-    proposal.status = ProposalStatus.REJECTED
-    proposal.decided_at = datetime.now(UTC)
-    proposal.decided_by = rejected_by
-    proposal.decision_reason = reason
-    store.move_proposal_to_decided(proposal)
-    audit.log_event(
-        store.kb_dir, event=f"proposal.{proposal.kind.value}.reject",
-        actor=rejected_by, object_ids=[proposal.id],
-        data={"reason": reason},
-    )
-    return proposal
+        return proposal
 
 
 def reject_auto_extracted(
@@ -698,29 +763,32 @@ def expire_one(
     expired_by: str = EXPIRE_ACTOR,
 ) -> Proposal:
     """Expire a single pending proposal (terminal reject + audit)."""
-    proposal = store.get_proposal(proposal_id)
-    if proposal.status != ProposalStatus.PENDING:
-        if (
-            proposal.status == ProposalStatus.REJECTED
-            and proposal.decision_reason == EXPIRE_REASON
-        ):
-            return proposal
-        raise ProposalError(
-            f"proposal {proposal_id} is {proposal.status.value}, not pending"
+    # Same decision lock as approve()/reject(): expiry is a reject, so it must
+    # not race a concurrent approve of the same proposal.
+    with audit.decision_lock(store.kb_dir):
+        proposal = store.get_proposal(proposal_id)
+        if proposal.status != ProposalStatus.PENDING:
+            if (
+                proposal.status == ProposalStatus.REJECTED
+                and proposal.decision_reason == EXPIRE_REASON
+            ):
+                return proposal
+            raise ProposalError(
+                f"proposal {proposal_id} is {proposal.status.value}, not pending"
+            )
+        proposal.status = ProposalStatus.REJECTED
+        proposal.decided_at = datetime.now(UTC)
+        proposal.decided_by = expired_by
+        proposal.decision_reason = EXPIRE_REASON
+        store.move_proposal_to_decided(proposal)
+        audit.log_event(
+            store.kb_dir,
+            event="proposal.expire",
+            actor=expired_by,
+            object_ids=[proposal.id],
+            data={"kind": proposal.kind.value},
         )
-    proposal.status = ProposalStatus.REJECTED
-    proposal.decided_at = datetime.now(UTC)
-    proposal.decided_by = expired_by
-    proposal.decision_reason = EXPIRE_REASON
-    store.move_proposal_to_decided(proposal)
-    audit.log_event(
-        store.kb_dir,
-        event="proposal.expire",
-        actor=expired_by,
-        object_ids=[proposal.id],
-        data={"kind": proposal.kind.value},
-    )
-    return proposal
+        return proposal
 
 
 def expire_pending(