diff --git a/aragora/cli/commands/dic20_decay_monitor.py b/aragora/cli/commands/dic20_decay_monitor.py index d26b17aeff..a71e391995 100644 --- a/aragora/cli/commands/dic20_decay_monitor.py +++ b/aragora/cli/commands/dic20_decay_monitor.py @@ -117,6 +117,7 @@ def cmd_decay_monitor(args: argparse.Namespace) -> int: file=sys.stderr, ) return 1 + units = [] signals = [] if manifests: from aragora.epistemic.proof_unit_model import load_proof_unit @@ -124,24 +125,40 @@ def cmd_decay_monitor(args: argparse.Namespace) -> int: for data in manifests: try: - signals.append( - evaluate_unit(load_proof_unit(data), claim_results=claim_results or None) - ) + unit = load_proof_unit(data) + units.append(unit) + signals.append(evaluate_unit(unit, claim_results=claim_results or None)) except Exception as exc: # noqa: BLE001 logger.warning("unit %s skipped: %s", data.get("code_unit_id", "?"), exc) + # Transitive impact set — exposes compute_decay_impact_set via the CLI. + # Active only when the caller sets transitive_impact=True in the namespace. + # No dependency edges are wired from manifests (single-hop impact only in + # this slice); multi-hop edge loading is DIC-20 follow-up scope. + transitive_impact_set: set[str] = set() + if getattr(args, "transitive_impact", False) and units: + from aragora.epistemic.constraint_graph import ProofUnitConstraintGraph + from aragora.epistemic.decay_monitor import compute_decay_impact_set + + failing_claim_ids: set[str] = { + r.claim_id + for s in signals + for r in s.reasons + if r.claim_id and r.kind in {"failed_claim", "stale_evidence", "verifier_error"} + } + graph = ProofUnitConstraintGraph(units) + transitive_impact_set = compute_decay_impact_set(graph, failing_claim_ids, transitive=True) + ts = datetime.now(timezone.utc).isoformat() if getattr(args, "json", False): - print( - json.dumps( - { - "generated_at": ts, - "total": len(signals), - "signals": [s.to_dict() for s in signals], - }, - indent=2, - ) - ) + out: dict = { + "generated_at": ts, + "total": len(signals), + "signals": [s.to_dict() for s in signals], + } + if transitive_impact_set: + out["transitive_impact_set"] = sorted(transitive_impact_set) + print(json.dumps(out, indent=2)) else: print(f"Decay monitor — {ts}\n{len(signals)} unit(s) evaluated\n") for s in signals: @@ -152,4 +169,8 @@ def cmd_decay_monitor(args: argparse.Namespace) -> int: print(f" [{r.kind}] {r.detail}") if not signals: print(" (no proof-unit manifests found)") + if transitive_impact_set: + print(f"\nTransitive impact ({len(transitive_impact_set)} unit(s)):") + for uid in sorted(transitive_impact_set): + print(f" {uid}") return 0 diff --git a/tests/cli/test_dic20_decay_monitor.py b/tests/cli/test_dic20_decay_monitor.py index 55fba4000f..81b6530812 100644 --- a/tests/cli/test_dic20_decay_monitor.py +++ b/tests/cli/test_dic20_decay_monitor.py @@ -166,3 +166,153 @@ def test_missing_claim_results_file_exits_1( monkeypatch.setenv(_FLAG, "1") assert cmd_decay_monitor(_ns(str(units_dir), claim_results=str(tmp_path / "nope.jsonl"))) == 1 assert "claim-results" in capsys.readouterr().err + + +# -- Transitive impact set -- + + +_UNIT_WITH_CLAIM_YAML = """\ +code_unit_id: test.unit.beta +version: "1.0" +claims: + - claim.beta.ok +decision_receipts: + - receipt-beta +decay_policy: + failed_claim: report_only + stale_evidence: report_only + unresolved_crux: report_only +""" + + +@pytest.fixture() +def units_dir_with_claim(tmp_path: Path) -> Path: + (tmp_path / "unit_beta.yaml").write_text(_UNIT_WITH_CLAIM_YAML, encoding="utf-8") + return tmp_path + + +def _ns_transitive( + units_dir: str, claim_results: str | None = None, json_out: bool = False +) -> argparse.Namespace: + return argparse.Namespace( + units_dir=units_dir, + claim_results=claim_results, + json=json_out, + transitive_impact=True, + ) + + +def test_transitive_impact_off_by_default_no_key_in_json( + monkeypatch, units_dir_with_claim: Path, tmp_path: Path, capsys +) -> None: + """Without transitive_impact flag, JSON output has no transitive_impact_set.""" + monkeypatch.setenv(_FLAG, "1") + cr = tmp_path / "cr.jsonl" + cr.write_text( + json.dumps({"claim_id": "claim.beta.ok", "status": "fail", "message": "test"}) + "\n", + encoding="utf-8", + ) + cmd_decay_monitor(_ns(str(units_dir_with_claim), claim_results=str(cr), json_out=True)) + out = json.loads(capsys.readouterr().out) + assert "transitive_impact_set" not in out + + +def test_transitive_impact_no_failures_no_key_in_json( + monkeypatch, units_dir_with_claim: Path, capsys +) -> None: + """With flag set but no failing claims, transitive_impact_set is omitted.""" + monkeypatch.setenv(_FLAG, "1") + cmd_decay_monitor(_ns_transitive(str(units_dir_with_claim), json_out=True)) + out = json.loads(capsys.readouterr().out) + assert "transitive_impact_set" not in out + + +def test_transitive_impact_failing_claim_includes_unit_id( + monkeypatch, units_dir_with_claim: Path, tmp_path: Path, capsys +) -> None: + """Failed claim marks its owning unit in transitive_impact_set.""" + monkeypatch.setenv(_FLAG, "1") + cr = tmp_path / "cr.jsonl" + cr.write_text( + json.dumps({"claim_id": "claim.beta.ok", "status": "fail", "message": "test"}) + "\n", + encoding="utf-8", + ) + cmd_decay_monitor( + _ns_transitive(str(units_dir_with_claim), claim_results=str(cr), json_out=True) + ) + out = json.loads(capsys.readouterr().out) + assert "transitive_impact_set" in out + assert "test.unit.beta" in out["transitive_impact_set"] + + +def test_transitive_impact_stale_claim_included( + monkeypatch, units_dir_with_claim: Path, tmp_path: Path, capsys +) -> None: + """Stale claim also contributes to transitive impact set.""" + monkeypatch.setenv(_FLAG, "1") + cr = tmp_path / "cr.jsonl" + cr.write_text( + json.dumps({"claim_id": "claim.beta.ok", "status": "stale", "message": "old"}) + "\n", + encoding="utf-8", + ) + cmd_decay_monitor( + _ns_transitive(str(units_dir_with_claim), claim_results=str(cr), json_out=True) + ) + out = json.loads(capsys.readouterr().out) + assert "transitive_impact_set" in out + assert "test.unit.beta" in out["transitive_impact_set"] + + +def test_transitive_impact_result_sorted(monkeypatch, tmp_path: Path, capsys) -> None: + """transitive_impact_set list is sorted for deterministic output.""" + monkeypatch.setenv(_FLAG, "1") + # Two units sharing a claim — both end up in impact set. + unit_z = """\ +code_unit_id: unit.z +version: "1.0" +claims: + - shared.claim +decision_receipts: ["r-z"] +decay_policy: + failed_claim: report_only + stale_evidence: report_only + unresolved_crux: report_only +""" + unit_a = """\ +code_unit_id: unit.a +version: "1.0" +claims: + - shared.claim +decision_receipts: ["r-a"] +decay_policy: + failed_claim: report_only + stale_evidence: report_only + unresolved_crux: report_only +""" + (tmp_path / "unit_z.yaml").write_text(unit_z, encoding="utf-8") + (tmp_path / "unit_a.yaml").write_text(unit_a, encoding="utf-8") + cr = tmp_path / "cr.jsonl" + cr.write_text( + json.dumps({"claim_id": "shared.claim", "status": "fail", "message": "x"}) + "\n", + encoding="utf-8", + ) + cmd_decay_monitor(_ns_transitive(str(tmp_path), claim_results=str(cr), json_out=True)) + out = json.loads(capsys.readouterr().out) + ids = out["transitive_impact_set"] + assert ids == sorted(ids) + + +def test_transitive_impact_text_output_contains_header( + monkeypatch, units_dir_with_claim: Path, tmp_path: Path, capsys +) -> None: + """Text output includes a 'Transitive impact' header when units are impacted.""" + monkeypatch.setenv(_FLAG, "1") + cr = tmp_path / "cr.jsonl" + cr.write_text( + json.dumps({"claim_id": "claim.beta.ok", "status": "fail", "message": "x"}) + "\n", + encoding="utf-8", + ) + cmd_decay_monitor(_ns_transitive(str(units_dir_with_claim), claim_results=str(cr))) + out = capsys.readouterr().out + assert "Transitive impact" in out + assert "test.unit.beta" in out