diff --git a/docs/multi-agent.md b/docs/multi-agent.md index 2ea14edd..ac051551 100644 --- a/docs/multi-agent.md +++ b/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 diff --git a/src/vouch/audit.py b/src/vouch/audit.py index a37bb1ca..e6a72681 100644 --- a/src/vouch/audit.py +++ b/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 diff --git a/src/vouch/index_db.py b/src/vouch/index_db.py index 29d6a859..52751b97 100644 --- a/src/vouch/index_db.py +++ b/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 diff --git a/src/vouch/proposals.py b/src/vouch/proposals.py index e66c1298..f2f0dc05 100644 --- a/src/vouch/proposals.py +++ b/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_() 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( diff --git a/src/vouch/storage.py b/src/vouch/storage.py index b2214d7b..ccadf205 100644 --- a/src/vouch/storage.py +++ b/src/vouch/storage.py @@ -23,12 +23,14 @@ from __future__ import annotations +import contextlib import hashlib import logging import os import re import sqlite3 import stat +import tempfile from pathlib import Path from typing import Any, TypeVar @@ -130,6 +132,43 @@ def sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() +def _atomic_write(path: Path, data: str, *, exclusive: bool = False) -> None: + """Write ``data`` to ``path`` durably: a crash leaves either the old bytes + or the complete new bytes, never a torn mix. + + Write to a temp sibling in the same directory (same filesystem, so the + placement is an atomic rename), ``fsync`` it, then put it in place -- + ``os.replace`` for an overwrite, or ``os.link`` (which raises + ``FileExistsError`` if the target already exists) to preserve the + create-only contract ``put_claim`` / ``put_page`` previously got from + ``open("x")``. The pre-placement ``fsync`` is what lets the audit append + that follows never attest to yaml power loss erased -- generalizing the + fsync-before-unlink that ``move_proposal_to_decided`` used to do inline. + + Mirrors the migration layer's ``migrations.rewriter.atomic_write_text``, + which imports from here, so the two stay separate by necessity.""" + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + tmp = Path(tmp_name) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + if exclusive: + os.link(tmp, path) # atomic create; FileExistsError if target present + else: + os.replace(tmp, path) # atomic overwrite; consumes tmp + except BaseException: + # Leave no temp behind on any failure, incl. KeyboardInterrupt. + with contextlib.suppress(FileNotFoundError): + os.unlink(tmp) + raise + # os.replace consumed tmp; os.link left it as a second link to drop. + with contextlib.suppress(FileNotFoundError): + os.unlink(tmp) + + def discover_root(start: Path | None = None) -> Path: """Walk up from `start` looking for a `.vouch` directory. @@ -351,7 +390,7 @@ def put_source( tags=tags or [], metadata=metadata or {}, ) - meta_path.write_text(_yaml_dump(src.model_dump(mode="json")), encoding="utf-8") + _atomic_write(meta_path, _yaml_dump(src.model_dump(mode="json"))) self._embed_and_store(kind="source", id=src.id, text=src.title or src.locator or "") return src @@ -458,8 +497,11 @@ def put_claim(self, claim: Claim) -> Claim: ) self._validate_claim_refs(claim) try: - with self._claim_path(claim.id).open("x", encoding="utf-8") as f: - f.write(_yaml_dump(claim.model_dump(mode="json"))) + _atomic_write( + self._claim_path(claim.id), + _yaml_dump(claim.model_dump(mode="json")), + exclusive=True, + ) except FileExistsError as e: raise ValueError( f"claim {claim.id} already exists -- use update_claim()" @@ -497,8 +539,10 @@ def update_claim(self, claim: Claim) -> Claim: # model validator can't catch (it has no KB access). Mirrors the # put_claim guard so the update path can't reintroduce the gap. self._validate_claim_refs(claim) - self._claim_path(claim.id).write_text( - _yaml_dump(claim.model_dump(mode="json")), encoding="utf-8") + _atomic_write( + self._claim_path(claim.id), + _yaml_dump(claim.model_dump(mode="json")), + ) self._embed_and_store(kind="claim", id=claim.id, text=claim.text) # Keep the FTS5 row in sync with the on-disk claim so lifecycle # mutations (archive, supersede, contradict, confirm) are reflected @@ -533,12 +577,13 @@ def put_page(self, page: Page) -> Page: if not (self._source_dir(sid) / "meta.yaml").exists(): raise ValueError(f"page {page.id} references unknown source {sid}") try: - # Explicit UTF-8: page bodies are user / agent prose and routinely - # contain non-ASCII (em-dashes, smart quotes, unicode in claims). - # The default text-mode encoding follows the locale (Latin-1 on a - # bare Linux container), which would mangle anything past 0x7F. - with self._page_path(page.id).open("x", encoding="utf-8") as f: - f.write(_serialize_page(page)) + # Explicit UTF-8 (in _atomic_write): page bodies are user / agent + # prose and routinely contain non-ASCII (em-dashes, smart quotes, + # unicode in claims). The default text-mode encoding follows the + # locale (Latin-1 on a bare Linux container), mangling past 0x7F. + _atomic_write( + self._page_path(page.id), _serialize_page(page), exclusive=True + ) except FileExistsError as e: raise ValueError( f"page {page.id} already exists -- choose a different slug" @@ -561,9 +606,7 @@ def update_page(self, page: Page) -> Page: """ if not self._page_path(page.id).exists(): raise ArtifactNotFoundError(f"page {page.id}") - self._page_path(page.id).write_text( - _serialize_page(page), encoding="utf-8" - ) + _atomic_write(self._page_path(page.id), _serialize_page(page)) self._embed_and_store( kind="page", id=page.id, text=f"{page.title}\n\n{page.body}", @@ -584,8 +627,11 @@ def list_pages(self) -> list[Page]: def put_entity(self, entity: Entity) -> Entity: try: - with self._entity_path(entity.id).open("x", encoding="utf-8") as f: - f.write(_yaml_dump(entity.model_dump(mode="json"))) + _atomic_write( + self._entity_path(entity.id), + _yaml_dump(entity.model_dump(mode="json")), + exclusive=True, + ) except FileExistsError as e: raise ValueError( f"entity {entity.id} already exists -- choose a different slug" @@ -633,8 +679,11 @@ def _validate_relation_refs(self, rel: Relation) -> None: def put_relation(self, rel: Relation) -> Relation: self._validate_relation_refs(rel) try: - with self._relation_path(rel.id).open("x", encoding="utf-8") as f: - f.write(_yaml_dump(rel.model_dump(mode="json"))) + _atomic_write( + self._relation_path(rel.id), + _yaml_dump(rel.model_dump(mode="json")), + exclusive=True, + ) except FileExistsError as e: raise ValueError( f"relation {rel.id} already exists -- choose a different slug" @@ -666,8 +715,9 @@ def put_relation_idempotent(self, rel: Relation) -> Relation: # the linked claim was subsequently archived or retracted. self._validate_relation_refs(rel) try: - with path.open("x", encoding="utf-8") as f: - f.write(_yaml_dump(rel.model_dump(mode="json"))) + _atomic_write( + path, _yaml_dump(rel.model_dump(mode="json")), exclusive=True + ) except FileExistsError: self._embed_and_store( kind="relation", id=rel.id, @@ -733,8 +783,11 @@ def put_evidence(self, ev: Evidence) -> Evidence: if not (self._source_dir(ev.source_id) / "meta.yaml").exists(): raise ValueError(f"evidence {ev.id} cites unknown source {ev.source_id}") try: - with self._evidence_path(ev.id).open("x", encoding="utf-8") as f: - f.write(_yaml_dump(ev.model_dump(mode="json"))) + _atomic_write( + self._evidence_path(ev.id), + _yaml_dump(ev.model_dump(mode="json")), + exclusive=True, + ) except FileExistsError as e: raise ValueError( f"evidence {ev.id} already exists -- choose a different slug" @@ -759,8 +812,11 @@ def list_evidence(self) -> list[Evidence]: def put_session(self, sess: Session) -> Session: try: - with self._session_path(sess.id).open("x", encoding="utf-8") as f: - f.write(_yaml_dump(sess.model_dump(mode="json"))) + _atomic_write( + self._session_path(sess.id), + _yaml_dump(sess.model_dump(mode="json")), + exclusive=True, + ) except FileExistsError as e: raise ValueError( f"session {sess.id} already exists -- choose a different id" @@ -773,8 +829,10 @@ def update_session(self, sess: Session) -> Session: # guard against duplicate ids, so updates need a separate path. if not self._session_path(sess.id).exists(): raise ArtifactNotFoundError(f"session {sess.id}") - self._session_path(sess.id).write_text( - _yaml_dump(sess.model_dump(mode="json")), encoding="utf-8") + _atomic_write( + self._session_path(sess.id), + _yaml_dump(sess.model_dump(mode="json")), + ) return sess def get_session(self, sid: str) -> Session: @@ -861,8 +919,11 @@ def _embed_and_store( def put_proposal(self, proposal: Proposal) -> Proposal: try: - with self._proposal_path(proposal.id).open("x", encoding="utf-8") as f: - f.write(_yaml_dump(proposal.model_dump(mode="json"))) + _atomic_write( + self._proposal_path(proposal.id), + _yaml_dump(proposal.model_dump(mode="json")), + exclusive=True, + ) except FileExistsError as e: raise ValueError( f"proposal {proposal.id} already exists -- choose a different id" @@ -878,32 +939,44 @@ def update_proposal(self, proposal: Proposal) -> Proposal: path = self._proposal_path(proposal.id) if not path.exists(): raise ArtifactNotFoundError(f"proposal {proposal.id}") - path.write_text( - _yaml_dump(proposal.model_dump(mode="json")), encoding="utf-8" - ) + _atomic_write(path, _yaml_dump(proposal.model_dump(mode="json"))) return proposal def get_proposal(self, proposal_id: str) -> Proposal: - for path in (self._proposal_path(proposal_id), self._decided_path(proposal_id)): + # decided/ wins over proposed/. move_proposal_to_decided writes decided/ + # then unlinks proposed/, so a crash between the two leaves both copies. + # The recorded decision is authoritative — returning the stale pending + # copy would let a recorded "no" resurface as pending and later flip to + # a durable "yes". + for path in (self._decided_path(proposal_id), self._proposal_path(proposal_id)): if path.exists(): return Proposal.model_validate(_yaml_load(path.read_text(encoding="utf-8"))) raise ArtifactNotFoundError(f"proposal {proposal_id}") def list_proposals(self, status: ProposalStatus | None = None) -> list[Proposal]: - out: list[Proposal] = [] - for sub in ("proposed", "decided"): + # decided/ wins over proposed/ (see get_proposal): a proposal present in + # both — the move_proposal_to_decided crash window — appears once, as its + # recorded decision, never twice. + by_id: dict[str, Proposal] = {} + for sub in ("decided", "proposed"): for p in sorted((self.kb_dir / sub).glob("*.yaml")): pr = _load_or_skip(p, Proposal, "proposal") - if pr is None: + if pr is None or pr.id in by_id: continue - if status is None or pr.status == status: - out.append(pr) - return out + by_id[pr.id] = pr + return sorted( + (pr for pr in by_id.values() if status is None or pr.status == status), + key=lambda pr: pr.id, + ) def move_proposal_to_decided(self, proposal: Proposal) -> None: src = self._proposal_path(proposal.id) dst = self._decided_path(proposal.id) - dst.write_text(_yaml_dump(proposal.model_dump(mode="json")), encoding="utf-8") + # _atomic_write fsyncs the decision before the rename that publishes it, + # so after it returns decided/ is durable. A crash may then leave both + # files (readers prefer decided) but must never leave neither, so the + # recorded decision has to be durable before proposed/ is unlinked. + _atomic_write(dst, _yaml_dump(proposal.model_dump(mode="json"))) if src.exists(): src.unlink() diff --git a/tests/test_index.py b/tests/test_index.py index e962dea0..54789afb 100644 --- a/tests/test_index.py +++ b/tests/test_index.py @@ -47,3 +47,14 @@ def test_index_via_approve(store: KBStore) -> None: approve(store, pr.id, approved_by="u") hits = index_db.search(store.kb_dir, "indexed") assert any(k == "claim" for k, *_ in hits) + + +def test_open_db_enables_wal_and_busy_timeout(store: KBStore) -> None: + """Fleet-safe sqlite: WAL keeps readers from blocking the single writer, + and a busy_timeout makes a contended writer wait instead of erroring at + once -- five concurrent agents on one KB is the dogfood config.""" + with index_db.open_db(store.kb_dir) as conn: + mode = conn.execute("PRAGMA journal_mode").fetchone()[0] + busy = conn.execute("PRAGMA busy_timeout").fetchone()[0] + assert mode.lower() == "wal" + assert busy >= 1000 diff --git a/tests/test_page_kinds.py b/tests/test_page_kinds.py index 267e2c65..67ae7930 100644 --- a/tests/test_page_kinds.py +++ b/tests/test_page_kinds.py @@ -10,7 +10,7 @@ from click.testing import CliRunner from vouch.cli import cli -from vouch.models import Page +from vouch.models import Page, PageStatus from vouch.page_kinds import PageKindError, load_page_kind_registry, validate_page from vouch.proposals import ProposalError, approve, propose_page from vouch.storage import KBStore @@ -51,6 +51,16 @@ def test_declared_kind_is_accepted(store: KBStore) -> None: assert store.get_page(page.id).metadata["attendees"] == ["alice-example", "bob-example"] +def test_approved_page_is_active_not_draft(store: KBStore) -> None: + """An approved page must be ACTIVE. Page defaults to DRAFT and the approve + path never stamped it, so reviewed pages read as draft forever — the wiki + served approved knowledge with a draft badge and STALE had no writer.""" + pr = propose_page(store, title="A page", body="b", proposed_by="agent") + page = approve(store, pr.id, approved_by="reviewer") + assert page.status is PageStatus.ACTIVE + assert store.get_page(page.id).status is PageStatus.ACTIVE + + def test_undeclared_kind_is_rejected(store: KBStore) -> None: with pytest.raises(ProposalError) as exc: propose_page(store, title="X", body="b", page_type="proposal", proposed_by="a") diff --git a/tests/test_storage.py b/tests/test_storage.py index ce775688..aafb12be 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -2,12 +2,14 @@ from __future__ import annotations +import os +import sqlite3 from pathlib import Path import pytest from pydantic import ValidationError -from vouch import audit, lifecycle +from vouch import audit, index_db, lifecycle, storage from vouch.models import ( Claim, ClaimStatus, @@ -709,6 +711,29 @@ def test_double_approve_rejected(store: KBStore) -> None: approve(store, pr.id, approved_by="u") +def test_approve_survives_index_write_failure( + store: KBStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + """state.db is derived, so a locked or broken FTS index must not brick the + durable approve. The claim lands, the decision is recorded, and the approve + event is logged. Without catch-and-degrade the claim file would exist while + the proposal stayed pending -- and a retry would hit the 'already exists' + guard, bricking the approve for good.""" + src = store.put_source(b"e") + pid = propose_claim(store, text="jwt", evidence=[src.id], proposed_by="a").id + + def _locked(*_a: object, **_k: object) -> None: + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(index_db, "index_claim", _locked) + result = approve(store, pid, approved_by="b") # must not raise + + assert store.get_claim(result.id).text == "jwt" + assert store.get_proposal(pid).status is ProposalStatus.APPROVED + audit_log = (store.kb_dir / "audit.log.jsonl").read_text(encoding="utf-8") + assert "proposal.claim.approve" in audit_log + + def test_approve_refuses_to_overwrite_existing_artifact(store: KBStore) -> None: # Regression for #11: approve() wrote the artifact before moving the # proposal to decided/. A crash between the two steps would leave the @@ -946,3 +971,220 @@ def test_list_pages_skips_unreadable_file(store: KBStore) -> None: pages = store.list_pages() assert [p.id for p in pages] == ["p-ok"] + + +# --- crash window: a recorded decision must win over a stale pending copy --- + + +def _simulate_decided_and_stale_proposed(store: KBStore, pid: str) -> None: + """Leave both a decided (APPROVED) copy and the pre-decision pending copy, + the exact state move_proposal_to_decided leaves if it crashes after writing + decided/ but before unlinking proposed/.""" + import yaml + + pending = store.get_proposal(pid) + approved = pending.model_copy( + update={"status": ProposalStatus.APPROVED, "decided_by": "b"} + ) + store.move_proposal_to_decided(approved) # decided/ = APPROVED, proposed/ gone + # re-create the stale pending copy that a crash would have left behind + store._proposal_path(pid).write_text( + yaml.safe_dump(pending.model_dump(mode="json")), encoding="utf-8" + ) + + +def test_get_proposal_prefers_decided_over_stale_proposed(store: KBStore) -> None: + src = store.put_source(b"e") + pid = propose_claim(store, text="x", evidence=[src.id], proposed_by="a").id + _simulate_decided_and_stale_proposed(store, pid) + # a recorded "no"/"yes" must not be masked by the leftover pending copy + assert store.get_proposal(pid).status is ProposalStatus.APPROVED + + +def test_list_proposals_dedups_preferring_decided(store: KBStore) -> None: + src = store.put_source(b"e") + pid = propose_claim(store, text="x", evidence=[src.id], proposed_by="a").id + _simulate_decided_and_stale_proposed(store, pid) + matching = [p for p in store.list_proposals() if p.id == pid] + assert len(matching) == 1 + assert matching[0].status is ProposalStatus.APPROVED + + +def test_approve_serializes_on_the_decision_lock( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """approve() must hold the per-KB decision lock across its read-check-write, + so a concurrent approve/reject on the same proposal cannot both proceed and + leave a durable claim under a REJECTED record.""" + import threading + + from vouch import audit + from vouch.proposals import approve, propose_claim + + monkeypatch.chdir(store.root) + src = store.put_source(b"e") + pid = propose_claim(store, text="x", evidence=[src.id], proposed_by="a").id + + lock_held = threading.Event() + release = threading.Event() + approve_done = threading.Event() + + def hold_lock() -> None: + with audit.decision_lock(store.kb_dir): + lock_held.set() + release.wait(timeout=5) + + def do_approve() -> None: + approve(store, pid, approved_by="b") + approve_done.set() + + holder = threading.Thread(target=hold_lock) + holder.start() + assert lock_held.wait(timeout=5) + + approver = threading.Thread(target=do_approve) + approver.start() + # while the decision lock is held elsewhere, approve must block + assert not approve_done.wait(timeout=0.5) + release.set() + assert approve_done.wait(timeout=5) + + holder.join(timeout=5) + approver.join(timeout=5) + assert store.get_proposal(pid).status is ProposalStatus.APPROVED + + +# --- durable artifact writes: tmp + fsync + os.replace -------------------- +# +# The fsynced audit log must never end up attesting to a claim/page/proposal +# whose yaml a crash left torn or erased. Every durable artifact write goes +# through _atomic_write: write a temp sibling, fsync it, then place it with an +# atomic rename (os.replace to overwrite, os.link to preserve the create-only +# contract of put_claim/put_page). A crash therefore leaves the old bytes or +# the complete new bytes -- never a half-written mix. + + +def test_atomic_write_overwrite_crash_preserves_old_bytes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +) -> None: + """A crash at the final rename leaves the previous good file intact and no + temp file behind -- the whole point of swapping through a tmp sibling.""" + p = tmp_path / "claim.yaml" + p.write_text("OLD", encoding="utf-8") + + def _boom(*_a: object, **_k: object) -> None: + raise RuntimeError("power loss") + + monkeypatch.setattr(os, "replace", _boom) + with pytest.raises(RuntimeError): + storage._atomic_write(p, "NEW") + + assert p.read_text(encoding="utf-8") == "OLD" + assert list(tmp_path.glob("*.tmp")) == [] + + +def test_atomic_write_fsyncs_data_before_rename( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +) -> None: + """Data is fsynced before the rename makes it visible, so the audit append + that follows can never attest to bytes power loss erased.""" + order: list[str] = [] + real_fsync, real_replace = os.fsync, os.replace + + def _fsync(fd: int) -> None: + order.append("fsync") + real_fsync(fd) + + def _replace(src: object, dst: object) -> None: + order.append("replace") + real_replace(src, dst) # type: ignore[arg-type] + + monkeypatch.setattr(os, "fsync", _fsync) + monkeypatch.setattr(os, "replace", _replace) + storage._atomic_write(tmp_path / "c.yaml", "data") + + assert order == ["fsync", "replace"] + assert (tmp_path / "c.yaml").read_text(encoding="utf-8") == "data" + + +def test_atomic_write_exclusive_rejects_existing_and_leaves_no_tmp( + tmp_path: Path, +) -> None: + """Exclusive mode keeps put_claim/put_page's create-only contract: an + existing target raises FileExistsError, unchanged, with no temp leaked.""" + p = tmp_path / "claim.yaml" + p.write_text("EXISTING", encoding="utf-8") + with pytest.raises(FileExistsError): + storage._atomic_write(p, "NEW", exclusive=True) + assert p.read_text(encoding="utf-8") == "EXISTING" + assert list(tmp_path.glob("*.tmp")) == [] + + +def test_atomic_write_exclusive_creates_when_absent(tmp_path: Path) -> None: + p = tmp_path / "sub" / "claim.yaml" + storage._atomic_write(p, "hello", exclusive=True) + assert p.read_text(encoding="utf-8") == "hello" + assert list(p.parent.glob("*.tmp")) == [] + + +def _spy_atomic_writes( + monkeypatch: pytest.MonkeyPatch, +) -> list[tuple[str, bool]]: + """Record (filename, exclusive) for every _atomic_write, still writing.""" + seen: list[tuple[str, bool]] = [] + real = storage._atomic_write + + def spy(path: Path, data: str, *, exclusive: bool = False) -> None: + seen.append((Path(path).name, exclusive)) + real(path, data, exclusive=exclusive) + + monkeypatch.setattr(storage, "_atomic_write", spy) + return seen + + +def test_put_and_update_claim_write_through_atomic_helper( + store: KBStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + src = store.put_source(b"e") + seen = _spy_atomic_writes(monkeypatch) + store.put_claim(Claim(id="c1", text="orig", evidence=[src.id])) + # create uses exclusive mode -- the "already exists" guard must survive. + assert ("c1.yaml", True) in seen + + seen.clear() + c = store.get_claim("c1") + c.text = "revised" + store.update_claim(c) + # overwrite uses the plain os.replace path. + assert ("c1.yaml", False) in seen + + +def test_put_and_update_page_write_through_atomic_helper( + store: KBStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + src = store.put_source(b"e") + store.put_claim(Claim(id="c1", text="t", evidence=[src.id])) + seen = _spy_atomic_writes(monkeypatch) + page = Page(id="p1", title="T", body="b", claims=["c1"]) + store.put_page(page) + assert ("p1.md", True) in seen + + seen.clear() + page.body = "b2" + store.update_page(page) + assert ("p1.md", False) in seen + + +def test_move_proposal_to_decided_writes_through_atomic_helper( + store: KBStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + src = store.put_source(b"e") + pid = propose_claim(store, text="x", evidence=[src.id], proposed_by="a").id + pending = store.get_proposal(pid) + seen = _spy_atomic_writes(monkeypatch) + decided = pending.model_copy( + update={"status": ProposalStatus.APPROVED, "decided_by": "b"} + ) + store.move_proposal_to_decided(decided) + # the recorded decision lands as a torn-proof atomic overwrite. + assert (f"{pid}.yaml", False) in seen