fix(engineering): V1-0 gap closures per Codex review

Codex audit of cbf9e03 surfaced two P1 gaps + one P2 scope concern,
all verified with code-level probes. Patches below.

P1: promote_entity did not re-check F-8 at status flip.
Legacy candidates with source_refs='[]' and hand_authored=0 can
exist from before V1-0 enforcement. promote_entity now raises
ValueError before flipping status so no F-8 violation can slip
into the active store through the promote path. Row stays
candidate on rejection. Symmetric error shape with the create
side.

P1: supersede_entity was missing the F-5 hook.
Plan calls for synchronous conflict detection on every
active-entity write path. Supersede creates a `supersedes`
relationship rooted at the `superseded_by` entity, which can
produce a conflict the detector should catch. Added
detect_conflicts_for_entity(superseded_by) call with fail-open
per conflict-model.md:256.

P2: backfill script --invalidate-instead was too broad.
Query included both active AND superseded rows; invalidating
superseded rows collapses audit history that V1-0 remediation
never intended to touch. Now --invalidate-instead scopes to
status='active' only. Default hand_authored-flag mode stays
broad since it's additive/non-destructive. Help text made the
destructive posture explicit.

Four new regression tests in test_v1_0_write_invariants.py:
- test_promote_rejects_legacy_candidate_without_provenance
- test_promote_accepts_candidate_flagged_hand_authored
- test_supersede_runs_conflict_detection_on_new_active
- test_supersede_hook_fails_open

Test count: 543 -> 547 (+4). Full suite green in 81.07s.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-22 14:49:42 -04:00
parent cbf9e03ab9
commit f16cd5272f
4 changed files with 180 additions and 4 deletions

View File

@@ -22,9 +22,12 @@ 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
@@ -122,6 +125,66 @@ def test_create_entity_with_empty_source_refs_list_is_treated_as_missing(tmp_dat
)
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 ----------
@@ -150,6 +213,71 @@ def test_active_create_runs_conflict_detection_hook(tmp_data_dir, monkeypatch):
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