H1 batch: GOV-001, AXW-020R/020A/020B/020C

app/adapters/machine_knowledge.py

@@ -60,6 +60,13 @@ def to_machine_knowledge_row(unit: MachineKnowledgeUnitV1) -> dict[str, Any]:
         raise ContractMappingError(
             "legacy machine knowledge row cannot represent approved governance"
         )
+    # AXW-020C/GOV-001 review: the legacy row has no scope column, so a scoped
+    # unit cannot be round-tripped losslessly. Fail closed instead of silently
+    # dropping the scope on the legacy path.
+    if unit.scope is not None:
+        raise ContractMappingError(
+            "legacy machine knowledge row cannot represent a scoped unit"
+        )
     expected_status = "legacy_active_unverified" if unit.legacy_active else "deprecated"
     if unit.lifecycle_status != expected_status:
         raise ContractMappingError(

app/contracts/v1.py

@@ -217,6 +217,10 @@ class MachineKnowledgeUnitV1(BaseModel):
     source_type: str = Field(min_length=1)
     source_id: str
     legacy_active: int = Field(ge=0, le=1)
+    # GOV-001: a unit may be scoped to a retrieval domain. None (default) means
+    # a generic rule visible to any retrieval; a set value means the unit is only
+    # visible to retrievals requesting that exact scope.
+    scope: str | None = None
     lifecycle_status: Literal[
         "candidate", "legacy_active_unverified", "approved", "deprecated"
     ]

app/evidence/anchor.py

@@ -0,0 +1,177 @@
+"""AXW-020C: EvidenceAnchor and IndexRevision.
+
+An EvidenceAnchor locates content within a source version — by page, block,
+character/region, or source revision. An IndexRevision records a rebuildable
+derived index (FTS/vector) that must never be presented as the source of
+truth; its rebuild count and source revision distinguish derived index from the
+original.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+def _stable_id(prefix: str, *parts: object) -> str:
+    payload = json.dumps(parts, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
+    return f"{prefix}_" + hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24]
+
+
+@dataclass(frozen=True)
+class EvidenceAnchor:
+    anchor_id: str
+    raw_sha256: str
+    source_revision: str
+    locator: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class IndexRevision:
+    revision_id: str
+    raw_sha256: str
+    index_name: str
+    source_revision: str
+    rebuild_count: int
+
+
+def build_evidence_anchor(
+    raw_sha256: str, source_revision: str, locator: dict[str, Any]
+) -> EvidenceAnchor:
+    """Build a stable EvidenceAnchor from a raw source hash, a source revision
+    and a locator (page/block/char-region). Empty locator or revision is
+    rejected: an anchor must always pin content to a specific source version.
+    """
+    if not raw_sha256:
+        raise ValueError("evidence anchor requires a raw source hash")
+    if not source_revision:
+        raise ValueError("evidence anchor requires a source revision")
+    if not locator:
+        raise ValueError("evidence anchor requires a non-empty locator")
+    anchor_id = _stable_id("ev", raw_sha256, source_revision, locator)
+    return EvidenceAnchor(
+        anchor_id=anchor_id,
+        raw_sha256=raw_sha256,
+        source_revision=source_revision,
+        locator=locator,
+    )
+
+
+_ANCHOR_SCHEMA = """
+CREATE TABLE IF NOT EXISTS evidence_anchors (
+    anchor_id TEXT PRIMARY KEY,
+    raw_sha256 TEXT NOT NULL,
+    source_revision TEXT NOT NULL,
+    locator_json TEXT NOT NULL
+);
+"""
+_INDEX_SCHEMA = """
+CREATE TABLE IF NOT EXISTS index_revisions (
+    revision_id TEXT PRIMARY KEY,
+    raw_sha256 TEXT NOT NULL,
+    index_name TEXT NOT NULL,
+    source_revision TEXT NOT NULL,
+    rebuild_count INTEGER NOT NULL
+);
+"""
+
+
+def store_evidence_anchor(db: str | Path, anchor: EvidenceAnchor) -> None:
+    with sqlite3.connect(Path(db)) as conn:
+        conn.executescript(_ANCHOR_SCHEMA)
+        conn.execute(
+            "INSERT OR REPLACE INTO evidence_anchors "
+            "(anchor_id, raw_sha256, source_revision, locator_json) VALUES (?,?,?,?)",
+            (
+                anchor.anchor_id,
+                anchor.raw_sha256,
+                anchor.source_revision,
+                json.dumps(anchor.locator, ensure_ascii=True, sort_keys=True),
+            ),
+        )
+        conn.commit()
+
+
+def resolve_evidence_anchor(db: str | Path, anchor_id: str) -> EvidenceAnchor | None:
+    with sqlite3.connect(Path(db)) as conn:
+        conn.row_factory = sqlite3.Row
+        conn.executescript(_ANCHOR_SCHEMA)
+        row = conn.execute(
+            "SELECT * FROM evidence_anchors WHERE anchor_id=?", (anchor_id,)
+        ).fetchone()
+    if row is None:
+        return None
+    return EvidenceAnchor(
+        anchor_id=row["anchor_id"],
+        raw_sha256=row["raw_sha256"],
+        source_revision=row["source_revision"],
+        locator=json.loads(row["locator_json"]),
+    )
+
+
+def mark_index_revision(
+    db: str | Path,
+    raw_sha256: str,
+    index_name: str,
+    source_revision: str,
+) -> IndexRevision:
+    """Record a rebuildable derived index revision. The index points at the raw
+    source hash so it can never be mistaken for the source of truth itself."""
+    if not raw_sha256:
+        raise ValueError("index revision requires a raw source hash")
+    if not index_name:
+        raise ValueError("index revision requires an index name")
+    revision_id = _stable_id("idx", raw_sha256, index_name)
+    revision = IndexRevision(
+        revision_id=revision_id,
+        raw_sha256=raw_sha256,
+        index_name=index_name,
+        source_revision=source_revision,
+        rebuild_count=1,
+    )
+    with sqlite3.connect(Path(db)) as conn:
+        conn.executescript(_INDEX_SCHEMA)
+        conn.execute(
+            "INSERT OR REPLACE INTO index_revisions "
+            "(revision_id, raw_sha256, index_name, source_revision, rebuild_count) VALUES (?,?,?,?,?)",
+            (
+                revision.revision_id,
+                revision.raw_sha256,
+                revision.index_name,
+                revision.source_revision,
+                revision.rebuild_count,
+            ),
+        )
+        conn.commit()
+    return revision
+
+
+def rebuild_index_revision(
+    db: str | Path, revision_id: str, new_source_revision: str
+) -> IndexRevision | None:
+    """Rebuild an existing derived index against a (possibly newer) source
+    revision, incrementing the rebuild count. Returns None if unknown."""
+    with sqlite3.connect(Path(db)) as conn:
+        conn.row_factory = sqlite3.Row
+        conn.executescript(_INDEX_SCHEMA)
+        row = conn.execute(
+            "SELECT * FROM index_revisions WHERE revision_id=?", (revision_id,)
+        ).fetchone()
+        if row is None:
+            return None
+        new_count = row["rebuild_count"] + 1
+        conn.execute(
+            "UPDATE index_revisions SET source_revision=?, rebuild_count=? WHERE revision_id=?",
+            (new_source_revision, new_count, revision_id),
+        )
+        conn.commit()
+        return IndexRevision(
+            revision_id=row["revision_id"],
+            raw_sha256=row["raw_sha256"],
+            index_name=row["index_name"],
+            source_revision=new_source_revision,
+            rebuild_count=new_count,
+        )

app/evidence/bundle.py

@@ -0,0 +1,86 @@
+"""AXW-024B: CrossValidation and EvidenceBundle.
+
+A bundle groups evidence about one claim with explicit relations
+(supports/refutes/qualifies), cross-source comparison, conflict detection, and
+a human-review gate. Caller-supplied bundles require review (fail-closed);
+invalid relations and unknown-evidence references are rejected.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from app.evidence.graph import EvidenceNode
+
+_VALID_KINDS = {"supports", "refutes", "qualifies"}
+
+
+class EvidenceBundleError(ValueError):
+    """Raised when an evidence bundle is invalid."""
+
+
+@dataclass(frozen=True)
+class BundleRelation:
+    evidence_id: str
+    kind: str
+
+
+@dataclass(frozen=True)
+class EvidenceBundle:
+    claim_id: str
+    evidence: list[EvidenceNode] = field(default_factory=list)
+    relations: list[BundleRelation] = field(default_factory=list)
+
+    def relation_for(self, evidence_id: str) -> str | None:
+        for r in self.relations:
+            if r.evidence_id == evidence_id:
+                return r.kind
+        return None
+
+    @property
+    def has_conflict(self) -> bool:
+        kinds = {r.kind for r in self.relations}
+        return "supports" in kinds and "refutes" in kinds
+
+    @property
+    def conflict_reason(self) -> str:
+        kinds = sorted({r.kind for r in self.relations})
+        return "mixed relations: " + ", ".join(kinds)
+
+
+def build_evidence_bundle(
+    *,
+    claim_id: str,
+    evidence: list[EvidenceNode],
+    relations: list[BundleRelation],
+) -> EvidenceBundle:
+    """Validate and build an EvidenceBundle.
+
+    Fail-closed: every relation must reference bundle evidence with a valid
+    kind, and a caller-supplied bundle (any evidence without human review)
+    must require review.
+    """
+    evidence_by_id = {e.evidence_id: e for e in evidence}
+    if not evidence_by_id:
+        raise EvidenceBundleError("bundle requires at least one evidence")
+
+    for relation in relations:
+        if relation.kind not in _VALID_KINDS:
+            raise EvidenceBundleError(f"invalid relation kind: {relation.kind}")
+        if relation.evidence_id not in evidence_by_id:
+            raise EvidenceBundleError(
+                f"relation references unknown evidence: {relation.evidence_id}"
+            )
+
+    for node in evidence:
+        if node.claim_id != claim_id:
+            raise EvidenceBundleError("evidence belongs to a different claim")
+        if node.provenance_status == "caller_supplied" and not node.requires_human_review:
+            raise EvidenceBundleError(
+                "caller-supplied bundle requires human review"
+            )
+
+    return EvidenceBundle(
+        claim_id=claim_id,
+        evidence=list(evidence),
+        relations=list(relations),
+    )

app/evidence/graph.py

@@ -0,0 +1,72 @@
+"""AXW-024A: Claim/Evidence core graph.
+
+A Claim may be backed by multiple Evidence items. Every Evidence node must be
+traceable to a source locator, a generation method, a review state and
+provenance. The validator rejects evidence that points at a different claim,
+that is caller-supplied without human review, or a claim with no evidence.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+class ClaimEvidenceError(ValueError):
+    """Raised when a Claim/Evidence graph is invalid."""
+
+
+@dataclass(frozen=True)
+class EvidenceNode:
+    evidence_id: str
+    claim_id: str
+    source_locator: str
+    generation: str
+    requires_human_review: bool
+    provenance_status: str
+
+
+@dataclass(frozen=True)
+class ClaimEvidenceGraph:
+    claim_id: str
+    claim_statement: str
+    evidence: list[EvidenceNode] = field(default_factory=list)
+
+    @property
+    def evidence_count(self) -> int:
+        return len(self.evidence)
+
+
+def _validate_evidence_node(node: EvidenceNode) -> None:
+    if not node.source_locator:
+        raise ClaimEvidenceError("evidence requires a source locator")
+    if not node.generation:
+        raise ClaimEvidenceError("evidence requires a generation method")
+    if node.provenance_status == "caller_supplied" and not node.requires_human_review:
+        raise ClaimEvidenceError("caller_supplied evidence requires human review")
+
+
+def build_claim_evidence_graph(
+    *, claim_id: str, claim_statement: str, evidence: list[EvidenceNode]
+) -> ClaimEvidenceGraph:
+    """Validate and build a Claim/Evidence graph.
+
+    Fail-closed: a claim must have at least one evidence, every evidence must
+    belong to the same claim, and each node must satisfy provenance/review
+    governance.
+    """
+    if not claim_id:
+        raise ClaimEvidenceError("claim requires an id")
+    if not claim_statement:
+        raise ClaimEvidenceError("claim requires a statement")
+    if not evidence:
+        raise ClaimEvidenceError("claim requires at least one evidence")
+
+    for node in evidence:
+        if node.claim_id != claim_id:
+            raise ClaimEvidenceError("evidence belongs to a different claim")
+        _validate_evidence_node(node)
+
+    return ClaimEvidenceGraph(
+        claim_id=claim_id,
+        claim_statement=claim_statement,
+        evidence=list(evidence),
+    )