feat(engineering): enforce V1-0 write invariants

2026-04-22 14:59:17 -04:00
parent 9ab5b3c9d8
commit 2712c5d2d0
11 changed files with 734 additions and 8 deletions
--- a/tests/test_v1_0_write_invariants.py
+++ b/tests/test_v1_0_write_invariants.py
@@ -0,0 +1,362 @@
+"""V1-0 write-time invariant tests.
+
+Covers the Engineering V1 completion plan Phase V1-0 acceptance:
+- F-1 shared-header fields: extractor_version + canonical_home + hand_authored
+  land in the entities table with working defaults
+- F-8 provenance enforcement: create_entity raises without source_refs
+  unless hand_authored=True
+- F-5 synchronous conflict-detection hook on any active-entity write
+  (create_entity with status="active" + the pre-existing promote_entity
+  path); fail-open per conflict-model.md:256
+- Q-3 "flag, never block": a conflict never 4xx-blocks the write
+- Q-4 partial trust: get_entities scope_only filters candidates out
+
+Plan: docs/plans/engineering-v1-completion-plan.md
+Spec: docs/architecture/engineering-v1-acceptance.md
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from atocore.engineering.service import (
+    EXTRACTOR_VERSION,
+    create_entity,
+    create_relationship,
+    get_entities,
+    get_entity,
+    init_engineering_schema,
+    promote_entity,
+    supersede_entity,
+)
+from atocore.models.database import get_connection, init_db
+
+
+# ---------- F-1: shared-header fields ----------
+
+
+def test_entity_row_has_shared_header_fields(tmp_data_dir):
+    init_db()
+    init_engineering_schema()
+    with get_connection() as conn:
+        cols = {row["name"] for row in conn.execute("PRAGMA table_info(entities)").fetchall()}
+    assert "extractor_version" in cols
+    assert "canonical_home" in cols
+    assert "hand_authored" in cols
+
+
+def test_created_entity_has_default_extractor_version_and_canonical_home(tmp_data_dir):
+    init_db()
+    init_engineering_schema()
+    e = create_entity(
+        entity_type="component",
+        name="Pivot Pin",
+        project="p04-gigabit",
+        source_refs=["test:fixture"],
+    )
+    assert e.extractor_version == EXTRACTOR_VERSION
+    assert e.canonical_home == "entity"
+    assert e.hand_authored is False
+
+    # round-trip through get_entity to confirm the row mapper returns
+    # the same values (not just the return-by-construct path)
+    got = get_entity(e.id)
+    assert got is not None
+    assert got.extractor_version == EXTRACTOR_VERSION
+    assert got.canonical_home == "entity"
+    assert got.hand_authored is False
+
+
+def test_explicit_extractor_version_is_persisted(tmp_data_dir):
+    init_db()
+    init_engineering_schema()
+    e = create_entity(
+        entity_type="decision",
+        name="Pick GF-PTFE pads",
+        project="p04-gigabit",
+        source_refs=["interaction:abc"],
+        extractor_version="custom-v2.3",
+    )
+    got = get_entity(e.id)
+    assert got.extractor_version == "custom-v2.3"
+
+
+# ---------- F-8: provenance enforcement ----------
+
+
+def test_create_entity_without_provenance_raises(tmp_data_dir):
+    init_db()
+    init_engineering_schema()
+    with pytest.raises(ValueError, match="source_refs required"):
+        create_entity(
+            entity_type="component",
+            name="No Provenance",
+            project="p04-gigabit",
+            hand_authored=False,  # explicit — bypasses the test-conftest auto-flag
+        )
+
+
+def test_create_entity_with_hand_authored_needs_no_source_refs(tmp_data_dir):
+    init_db()
+    init_engineering_schema()
+    e = create_entity(
+        entity_type="component",
+        name="Human Entry",
+        project="p04-gigabit",
+        hand_authored=True,
+    )
+    assert e.hand_authored is True
+    got = get_entity(e.id)
+    assert got.hand_authored is True
+    # source_refs stays empty — the hand_authored flag IS the provenance
+    assert got.source_refs == []
+
+
+def test_create_entity_with_empty_source_refs_list_is_treated_as_missing(tmp_data_dir):
+    init_db()
+    init_engineering_schema()
+    with pytest.raises(ValueError, match="source_refs required"):
+        create_entity(
+            entity_type="component",
+            name="Empty Refs",
+            project="p04-gigabit",
+            source_refs=[],
+            hand_authored=False,
+        )
+
+
+def test_promote_rejects_legacy_candidate_without_provenance(tmp_data_dir):
+    """Regression (Codex V1-0 probe): candidate rows can exist in the DB
+    from before V1-0 enforcement (or from paths that bypass create_entity).
+    promote_entity must re-check the invariant and refuse to flip a
+    no-provenance candidate to active. Without this check, the active
+    store can leak F-8 violations in from legacy data."""
+    init_db()
+    init_engineering_schema()
+
+    # Simulate a pre-V1-0 candidate by inserting directly into the table,
+    # bypassing the service-layer invariant. Real legacy rows look exactly
+    # like this: empty source_refs, hand_authored=0.
+    import uuid as _uuid
+    entity_id = str(_uuid.uuid4())
+    with get_connection() as conn:
+        conn.execute(
+            "INSERT INTO entities (id, entity_type, name, project, "
+            "description, properties, status, confidence, source_refs, "
+            "extractor_version, canonical_home, hand_authored, "
+            "created_at, updated_at) "
+            "VALUES (?, 'component', 'Legacy Orphan', 'p04-gigabit', "
+            "'', '{}', 'candidate', 1.0, '[]', '', 'entity', 0, "
+            "CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)",
+            (entity_id,),
+        )
+
+    with pytest.raises(ValueError, match="source_refs required"):
+        promote_entity(entity_id)
+
+    # And the row stays a candidate — no half-transition.
+    got = get_entity(entity_id)
+    assert got is not None
+    assert got.status == "candidate"
+
+
+def test_promote_accepts_candidate_flagged_hand_authored(tmp_data_dir):
+    """The other side of the promote re-check: hand_authored=1 with
+    empty source_refs still lets promote succeed, matching
+    create_entity's symmetry."""
+    init_db()
+    init_engineering_schema()
+
+    import uuid as _uuid
+    entity_id = str(_uuid.uuid4())
+    with get_connection() as conn:
+        conn.execute(
+            "INSERT INTO entities (id, entity_type, name, project, "
+            "description, properties, status, confidence, source_refs, "
+            "extractor_version, canonical_home, hand_authored, "
+            "created_at, updated_at) "
+            "VALUES (?, 'component', 'Hand Authored Candidate', "
+            "'p04-gigabit', '', '{}', 'candidate', 1.0, '[]', '', "
+            "'entity', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)",
+            (entity_id,),
+        )
+
+    assert promote_entity(entity_id) is True
+    assert get_entity(entity_id).status == "active"
+
+
+# ---------- F-5: synchronous conflict-detection hook ----------
+
+
+def test_active_create_runs_conflict_detection_hook(tmp_data_dir, monkeypatch):
+    """status=active writes trigger detect_conflicts_for_entity."""
+    init_db()
+    init_engineering_schema()
+
+    called_with: list[str] = []
+
+    def _fake_detect(entity_id: str):
+        called_with.append(entity_id)
+        return []
+
+    import atocore.engineering.conflicts as conflicts_mod
+    monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _fake_detect)
+
+    e = create_entity(
+        entity_type="component",
+        name="Active With Hook",
+        project="p04-gigabit",
+        source_refs=["test:hook"],
+        status="active",
+    )
+
+    assert called_with == [e.id]
+
+
+def test_supersede_runs_conflict_detection_on_new_active(tmp_data_dir, monkeypatch):
+    """Regression (Codex V1-0 probe): per plan's 'every active-entity
+    write path', supersede_entity must trigger synchronous conflict
+    detection. The subject is the `superseded_by` entity — the one
+    whose graph state just changed because a new `supersedes` edge was
+    rooted at it."""
+    init_db()
+    init_engineering_schema()
+
+    old = create_entity(
+        entity_type="component",
+        name="Old Pad",
+        project="p04-gigabit",
+        source_refs=["test:old"],
+        status="active",
+    )
+    new = create_entity(
+        entity_type="component",
+        name="New Pad",
+        project="p04-gigabit",
+        source_refs=["test:new"],
+        status="active",
+    )
+
+    called_with: list[str] = []
+
+    def _fake_detect(entity_id: str):
+        called_with.append(entity_id)
+        return []
+
+    import atocore.engineering.conflicts as conflicts_mod
+    monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _fake_detect)
+
+    assert supersede_entity(old.id, superseded_by=new.id) is True
+
+    # The detector fires on the `superseded_by` entity — the one whose
+    # edges just grew a new `supersedes` relationship.
+    assert new.id in called_with
+
+
+def test_supersede_hook_fails_open(tmp_data_dir, monkeypatch):
+    """Supersede must survive a broken detector per Q-3 flag-never-block."""
+    init_db()
+    init_engineering_schema()
+
+    old = create_entity(
+        entity_type="component", name="Old2", project="p04-gigabit",
+        source_refs=["test:old"], status="active",
+    )
+    new = create_entity(
+        entity_type="component", name="New2", project="p04-gigabit",
+        source_refs=["test:new"], status="active",
+    )
+
+    def _boom(entity_id: str):
+        raise RuntimeError("synthetic detector failure")
+
+    import atocore.engineering.conflicts as conflicts_mod
+    monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _boom)
+
+    # The supersede still succeeds despite the detector blowing up.
+    assert supersede_entity(old.id, superseded_by=new.id) is True
+    assert get_entity(old.id).status == "superseded"
+
+
+def test_candidate_create_does_not_run_conflict_hook(tmp_data_dir, monkeypatch):
+    """status=candidate writes do NOT trigger detection — the hook is
+    for active rows only, per V1-0 scope. Candidates are checked at
+    promote time."""
+    init_db()
+    init_engineering_schema()
+
+    called: list[str] = []
+
+    def _fake_detect(entity_id: str):
+        called.append(entity_id)
+        return []
+
+    import atocore.engineering.conflicts as conflicts_mod
+    monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _fake_detect)
+
+    create_entity(
+        entity_type="component",
+        name="Candidate No Hook",
+        project="p04-gigabit",
+        source_refs=["test:cand"],
+        status="candidate",
+    )
+
+    assert called == []
+
+
+# ---------- Q-3: flag, never block ----------
+
+
+def test_conflict_detector_failure_does_not_block_write(tmp_data_dir, monkeypatch):
+    """Per conflict-model.md:256: detection errors must not fail the
+    write. The entity is still created; only a warning is logged."""
+    init_db()
+    init_engineering_schema()
+
+    def _boom(entity_id: str):
+        raise RuntimeError("synthetic detector failure")
+
+    import atocore.engineering.conflicts as conflicts_mod
+    monkeypatch.setattr(conflicts_mod, "detect_conflicts_for_entity", _boom)
+
+    # The write still succeeds — no exception propagates.
+    e = create_entity(
+        entity_type="component",
+        name="Hook Fails Open",
+        project="p04-gigabit",
+        source_refs=["test:failopen"],
+        status="active",
+    )
+    assert get_entity(e.id) is not None
+
+
+# ---------- Q-4 (partial): trust-hierarchy — scope_only filters candidates ----------
+
+
+def test_scope_only_active_does_not_return_candidates(tmp_data_dir):
+    """V1-0 partial Q-4: active-scoped listing never returns candidates.
+    Full trust-hierarchy coverage (no-auto-project-state, etc.) ships in
+    V1-E per plan."""
+    init_db()
+    init_engineering_schema()
+
+    active = create_entity(
+        entity_type="component",
+        name="Active Alpha",
+        project="p04-gigabit",
+        source_refs=["test:alpha"],
+        status="active",
+    )
+    candidate = create_entity(
+        entity_type="component",
+        name="Candidate Beta",
+        project="p04-gigabit",
+        source_refs=["test:beta"],
+        status="candidate",
+    )
+
+    listed = get_entities(project="p04-gigabit", status="active", scope_only=True)
+    ids = {e.id for e in listed}
+    assert active.id in ids
+    assert candidate.id not in ids