[DIC-20] transitive-impact-set CLI flag for aragora decay-monitor

aragora/cli/commands/dic20_decay_monitor.py

@@ -117,31 +117,48 @@ 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
         from aragora.epistemic.decay_monitor import evaluate_unit
 
         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

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