"""Engine-level correction and refutation tests, including cascade invalidation through a second-order deriver and sticky-refutation semantics across re-runs. A small in-test `ChainDeriver` produces second-order claims from every first-order claim, guaranteeing that the derivation graph has depth >= 2 so that cascades are exercised deterministically regardless of how the default derivers happen to layer.""" from __future__ import annotations import json import pytest from mnema.derive.confidence import ConfidenceFactor from mnema.derive.derivers.base import Deriver from mnema.derive.engine import DerivationEngine, default_derivers from mnema.derive.model import CandidateClaim, ClaimStatus CHAIN_PREDICATE = "test/second_order_summary" def _value_key(value) -> str: return json.dumps(value, sort_keys=True, default=str) class ChainDeriver(Deriver): """Derives one second-order claim from every active first-order claim.""" name = "test-chain" def derive(self, ctx): candidates = [] for claim in ctx.claims: if claim.predicate == CHAIN_PREDICATE: continue # never chain on our own output candidates.append( CandidateClaim( subject=claim.subject, predicate=CHAIN_PREDICATE, value={ "summarizes": claim.predicate, "of": _value_key(claim.value), }, inputs=[claim.claim_id], factors=[ ConfidenceFactor( name="upstream-claim", weight=0.8, detail="second-order summary used by cascade tests", ) ], ) ) return candidates @pytest.fixture() def chained(op_log, keypair, sample_evidence) -> DerivationEngine: engine = DerivationEngine( log=op_log, keypair=keypair, derivers=list(default_derivers()) + [ChainDeriver()], ) engine.ingest(sample_evidence) engine.run() engine.run() # settle second-order claims even without internal fixed-point return engine def _pick_base_with_descendants(engine): """Return (first-order claim, its transitive descendant claim-ids).""" for claim in engine.claims(): if claim.predicate == CHAIN_PREDICATE: continue descendants = set(engine.graph.descendants(claim.claim_id)) if descendants: return claim, descendants pytest.fail("expected at least one claim with chained descendants") def test_chain_claims_exist_and_record_claim_inputs(chained): engine = chained chain_claims = [c for c in engine.claims() if c.predicate == CHAIN_PREDICATE] assert chain_claims, "ChainDeriver must produce second-order claims" claim_ids = {c.claim_id for c in engine.claims()} for c in chain_claims: inputs = set(engine.graph.inputs(c.claim_id)) assert inputs & claim_ids, ( "second-order claims must record claim ids (not just evidence) as inputs" ) def test_refutation_invalidates_all_descendants(chained): engine = chained base, descendants = _pick_base_with_descendants(engine) report = engine.refute(base.claim_id, reason="user says this is wrong") invalidated = set(report.invalidated) assert descendants <= invalidated, ( f"descendants not cascaded: {sorted(descendants - invalidated)}" ) assert engine.graph.status(base.claim_id) == ClaimStatus.REFUTED for d in descendants: assert engine.graph.status(d) == ClaimStatus.INVALIDATED, ( f"descendant {d} should be INVALIDATED after upstream refutation" ) active = {c.claim_id for c in engine.claims()} assert base.claim_id not in active assert not (descendants & active) def test_refutation_is_sticky_across_reruns(chained): engine = chained base, _descendants = _pick_base_with_descendants(engine) triple = (base.subject, base.predicate, _value_key(base.value)) engine.refute(base.claim_id, reason="not true about me") engine.run() engine.run() active_triples = { (c.subject, c.predicate, _value_key(c.value)) for c in engine.claims() } assert triple not in active_triples, ( "a refuted claim must not be re-derived from the same evidence" ) # Nothing downstream may silently rebuild on the refuted claim either. for c in engine.claims(): assert base.claim_id not in set(engine.graph.inputs(c.claim_id)), ( f"active claim {c.claim_id} still depends on refuted {base.claim_id}" ) def test_refutation_appends_signed_operation(chained, op_log): engine = chained base, _ = _pick_base_with_descendants(engine) engine.refute(base.claim_id, reason="audit trail check") kinds = {op.kind for op in op_log} assert "refutation" in kinds, "refutations must be persisted to the operation log" def _corrected_value(value): if isinstance(value, dict): out = dict(value) out["corrected_by_user"] = True return out return {"corrected_by_user": True, "original": value} def test_correction_replaces_claim_and_invalidates_descendants(chained): engine = chained base, descendants = _pick_base_with_descendants(engine) new_value = _corrected_value(base.value) report = engine.correct(base.claim_id, value=new_value, reason="user correction") replacement = report.replacement assert replacement.claim_id != base.claim_id assert replacement.subject == base.subject assert replacement.predicate == base.predicate assert _value_key(replacement.value) == _value_key(new_value) assert 0.0 < float(replacement.confidence) <= 1.0 active = {c.claim_id for c in engine.claims()} assert base.claim_id not in active, "corrected claim must no longer be active" assert replacement.claim_id in active, "replacement claim must be active" for d in descendants: assert engine.graph.status(d) != ClaimStatus.ACTIVE, ( f"descendant {d} should be invalidated by upstream correction" ) def test_correction_rederives_downstream_from_replacement(chained): engine = chained base, _descendants = _pick_base_with_descendants(engine) new_value = _corrected_value(base.value) report = engine.correct(base.claim_id, value=new_value, reason="fix it") replacement = report.replacement engine.run() engine.run() rederived = [ c for c in engine.claims() if c.predicate == CHAIN_PREDICATE and replacement.claim_id in set(engine.graph.inputs(c.claim_id)) ] assert rederived, ( "downstream claims must be re-derived from the corrected replacement claim" ) # And nothing active may still cite the superseded original. for c in engine.claims(): assert base.claim_id not in set(engine.graph.inputs(c.claim_id)) def test_correction_appends_signed_operation(chained, op_log): engine = chained base, _ = _pick_base_with_descendants(engine) engine.correct(base.claim_id, value=_corrected_value(base.value), reason="log it") kinds = {op.kind for op in op_log} assert "correction" in kinds, "corrections must be persisted to the operation log" def test_correction_does_not_resurrect_original_on_rerun(chained): engine = chained base, _ = _pick_base_with_descendants(engine) original_triple = (base.subject, base.predicate, _value_key(base.value)) engine.correct(base.claim_id, value=_corrected_value(base.value), reason="r") engine.run() engine.run() active_triples = { (c.subject, c.predicate, _value_key(c.value)) for c in engine.claims() } assert original_triple not in active_triples, ( "the pre-correction value must not be re-derived over the user's correction" )