diff --git a/core/contemplation/__main__.py b/core/contemplation/__main__.py index 2a5beba0..80f49765 100644 --- a/core/contemplation/__main__.py +++ b/core/contemplation/__main__.py @@ -6,9 +6,17 @@ import sys from pathlib import Path from core.contemplation.runner import ( + contemplate_contradiction_reports, contemplate_frontier_reports, write_contemplation_run, ) +from teaching.discovery_sink import DiscoveryMonthlyFileSink + + +_LANE_RUNNERS = { + "frontier_compare": contemplate_frontier_reports, + "contradiction_detection": contemplate_contradiction_reports, +} def build_parser() -> argparse.ArgumentParser: @@ -20,7 +28,13 @@ def build_parser() -> argparse.ArgumentParser: "reports", nargs="+", type=Path, - help="frontier_compare JSON report path(s) to contemplate.", + help="Report JSON path(s) to contemplate. All paths must share --lane.", + ) + parser.add_argument( + "--lane", + choices=sorted(_LANE_RUNNERS.keys()), + default="frontier_compare", + help="Evidence lane the reports belong to (default: frontier_compare).", ) parser.add_argument( "--pack-id", @@ -38,18 +52,38 @@ def build_parser() -> argparse.ArgumentParser: "--report", type=Path, default=None, - help="Optional output path for the contemplation run JSON.", + help="Optional output path for the contemplation run JSON blob.", + ) + parser.add_argument( + "--sink-root", + type=Path, + default=None, + help=( + "Optional append-only JSONL sink root. When set each finding is " + "emitted as one canonical JSONL line via " + "teaching.discovery_sink.DiscoveryMonthlyFileSink at " + "//.jsonl alongside discovery candidates." + ), ) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) - run = contemplate_frontier_reports( - args.reports, - pack_ids=tuple(args.pack_id or ()), - notes=tuple(args.note or ()), + runner = _LANE_RUNNERS[args.lane] + sink = ( + DiscoveryMonthlyFileSink(args.sink_root) if args.sink_root is not None else None ) + try: + run = runner( + args.reports, + pack_ids=tuple(args.pack_id or ()), + notes=tuple(args.note or ()), + sink=sink, + ) + finally: + if sink is not None: + sink.close() if args.report is not None: write_contemplation_run(run, args.report) print(json.dumps(run.as_dict(), ensure_ascii=False, indent=2, sort_keys=True)) diff --git a/core/contemplation/miners/__init__.py b/core/contemplation/miners/__init__.py index 5540821e..77fed8cf 100644 --- a/core/contemplation/miners/__init__.py +++ b/core/contemplation/miners/__init__.py @@ -1,5 +1,9 @@ """Read-only contemplation miners.""" +from .contradiction_detection import mine_contradiction_detection_report from .frontier_compare import mine_frontier_compare_report -__all__ = ["mine_frontier_compare_report"] +__all__ = [ + "mine_contradiction_detection_report", + "mine_frontier_compare_report", +] diff --git a/core/contemplation/miners/contradiction_detection.py b/core/contemplation/miners/contradiction_detection.py new file mode 100644 index 00000000..75c97e91 --- /dev/null +++ b/core/contemplation/miners/contradiction_detection.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from core.contemplation.schema import ( + ContemplationEvidenceRef, + ContemplationFinding, + FindingKind, +) + + +_MISSED = "missed_contradiction" +_FALSE_FLAG = "false_contradiction_flag" + + +def _case_iter(report: dict[str, Any]) -> tuple[dict[str, Any], ...]: + return tuple(c for c in report.get("cases", ()) if isinstance(c, dict)) + + +def _failure_mode(case: dict[str, Any]) -> str | None: + """Classify a failed contradiction-detection case. + + Returns the predicate for the emitted finding, or ``None`` if the + case is not a failure worth flagging. + """ + if bool(case.get("passed", True)): + return None + kind = str(case.get("kind", "")) + if kind == "paired_contradiction": + return _MISSED + if kind == "paired_consistent": + return _FALSE_FLAG + return None + + +def _evidence_summary(case: dict[str, Any]) -> str: + """Compact, deterministic one-line evidence summary. + + Captures the lane's quantitative signals so a reviewer can decide + without re-running the lane. Keys are sorted to keep the summary + stable across runs. + """ + keys = ("kind", "flagged", "contested", "versor_delta", "versor_spike") + parts = [f"{k}={case[k]}" for k in keys if k in case] + return ";".join(parts) if parts else "case failed" + + +def mine_contradiction_detection_report( + report_path: str | Path, + *, + substrate_hash: str, +) -> tuple[ContemplationFinding, ...]: + """Convert failed contradiction-detection cases into speculative findings. + + Two failure modes are surfaced as distinct predicates: + + - ``missed_contradiction`` — a ``paired_contradiction`` case + the lane failed to flag (a real contradiction slipped through). + - ``false_contradiction_flag`` — a ``paired_consistent`` case + the lane wrongly flagged (the detector fired on consistent text). + + Both warrant operator attention but call for different repairs, so + the predicate split is load-bearing — not cosmetic. + + Read-only: parses ``report_path`` and returns immutable findings. + Never writes packs, teaching corpora, or runtime state. + """ + path = Path(report_path) + report = json.loads(path.read_text(encoding="utf-8")) + findings: list[ContemplationFinding] = [] + source_id = str(path) + lane = str(report.get("lane", "contradiction_detection")) + + for case in _case_iter(report): + predicate = _failure_mode(case) + if predicate is None: + continue + case_id = str(case.get("id", "unknown_case")) + evidence = ContemplationEvidenceRef( + source_type="contradiction_detection_report", + source_id=source_id, + pointer=f"lane={lane};case={case_id}", + summary=_evidence_summary(case), + ) + proposed_action = ( + "Inspect the paired-contradiction probe that slipped through; " + "either tighten the detector's versor-delta threshold or add a " + "focused regression for this case." + if predicate == _MISSED + else "Inspect the paired-consistent probe the detector wrongly " + "flagged; either loosen the threshold or add a counter-example " + "regression that pins the consistent reading." + ) + findings.append( + ContemplationFinding( + kind=FindingKind.CONTRADICTION, + subject=f"{lane}/{case_id}", + predicate=predicate, + object=None, + evidence_refs=(evidence,), + proposed_action=proposed_action, + substrate_hash=substrate_hash, + ) + ) + + return tuple(findings) + + +__all__ = ["mine_contradiction_detection_report"] diff --git a/core/contemplation/runner.py b/core/contemplation/runner.py index 881e1435..584312c0 100644 --- a/core/contemplation/runner.py +++ b/core/contemplation/runner.py @@ -5,9 +5,17 @@ import json from pathlib import Path from typing import Iterable +from core.contemplation.miners.contradiction_detection import ( + mine_contradiction_detection_report, +) from core.contemplation.miners.frontier_compare import mine_frontier_compare_report -from core.contemplation.schema import ContemplationFinding, ContemplationRun +from core.contemplation.schema import ( + ContemplationFinding, + ContemplationRun, + format_contemplation_finding_jsonl, +) from core.contemplation.snapshot import ContemplationSubstrate +from teaching.discovery_sink import DiscoveryCandidateSink def _config_hash(payload: dict[str, object]) -> str: @@ -20,17 +28,50 @@ def _config_hash(payload: dict[str, object]) -> str: return hashlib.sha256(encoded.encode("utf-8")).hexdigest() +def _emit_findings( + findings: Iterable[ContemplationFinding], + sink: DiscoveryCandidateSink | None, +) -> None: + """Stream each finding through the shared sink protocol when set. + + No-op when *sink* is ``None`` — preserves the existing "build a + ``ContemplationRun`` blob" path for callers that want a single + in-memory result. + + When a sink is supplied, each finding is emitted as one canonical + JSONL line via :func:`format_contemplation_finding_jsonl`. This + is the unification point with the discovery candidate stream + (ADR-0055 Phase B sinks): both flow through the same + ``DiscoveryCandidateSink`` protocol, both land in append-only + monthly JSONL files when paired with + :class:`teaching.discovery_sink.DiscoveryMonthlyFileSink`. + + Sink errors are NOT swallowed — see ADR-0055 fail-fast contract. + """ + if sink is None: + return + for finding in findings: + sink.emit(format_contemplation_finding_jsonl(finding)) + + def contemplate_frontier_reports( report_paths: Iterable[str | Path], *, pack_ids: Iterable[str] = (), notes: Iterable[str] = (), + sink: DiscoveryCandidateSink | None = None, ) -> ContemplationRun: """Run ADR-0080 Phase 1 over explicit frontier-compare reports. The runner is read-only. It does not discover files implicitly, does not mutate packs, does not write teaching examples, and does not promote any finding beyond SPECULATIVE. + + When *sink* is supplied each finding is also emitted as one + canonical JSONL line via the shared + :class:`teaching.discovery_sink.DiscoveryCandidateSink` protocol, + so contemplation findings flow into the same append-only stream + discovery candidates use. """ paths = tuple(Path(p) for p in report_paths) @@ -47,6 +88,7 @@ def contemplate_frontier_reports( substrate_hash=substrate.substrate_hash, ) ) + _emit_findings(findings, sink) config_hash = _config_hash( { "runner": "contemplate_frontier_reports", @@ -62,6 +104,51 @@ def contemplate_frontier_reports( ) +def contemplate_contradiction_reports( + report_paths: Iterable[str | Path], + *, + pack_ids: Iterable[str] = (), + notes: Iterable[str] = (), + sink: DiscoveryCandidateSink | None = None, +) -> ContemplationRun: + """Run ADR-0080 Phase 1 over explicit contradiction-detection reports. + + Mirrors :func:`contemplate_frontier_reports` for the + ``evals/contradiction_detection`` lane. Same read-only guarantees, + same SPECULATIVE-only finding contract, separate runner so the + config hash records which lane was contemplated. + """ + + paths = tuple(Path(p) for p in report_paths) + substrate = ContemplationSubstrate.from_report_paths( + paths, + pack_ids=tuple(pack_ids), + notes=tuple(notes), + ) + findings: list[ContemplationFinding] = [] + for path in paths: + findings.extend( + mine_contradiction_detection_report( + path, + substrate_hash=substrate.substrate_hash, + ) + ) + _emit_findings(findings, sink) + config_hash = _config_hash( + { + "runner": "contemplate_contradiction_reports", + "report_paths": [str(p) for p in paths], + "pack_ids": tuple(sorted(set(pack_ids))), + "notes": tuple(notes), + } + ) + return ContemplationRun( + substrate_hash=substrate.substrate_hash, + config_hash=config_hash, + findings=tuple(findings), + ) + + def write_contemplation_run(run: ContemplationRun, path: str | Path) -> None: target = Path(path) target.parent.mkdir(parents=True, exist_ok=True) @@ -71,4 +158,8 @@ def write_contemplation_run(run: ContemplationRun, path: str | Path) -> None: ) -__all__ = ["contemplate_frontier_reports", "write_contemplation_run"] +__all__ = [ + "contemplate_contradiction_reports", + "contemplate_frontier_reports", + "write_contemplation_run", +] diff --git a/core/contemplation/schema.py b/core/contemplation/schema.py index 88ede714..a8f26263 100644 --- a/core/contemplation/schema.py +++ b/core/contemplation/schema.py @@ -40,9 +40,40 @@ def _sha256_16(payload: dict[str, Any]) -> str: return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()[:16] +# BOUNDARY (vs teaching/discovery.py:EvidencePointer) +# --------------------------------------------------------------------------- +# ``EvidencePointer`` (teaching/discovery.py) and ``ContemplationEvidenceRef`` +# below intentionally remain separate types. They have *different +# semantics*, not just different shapes: +# +# - ``EvidencePointer.source`` is constrained to +# ``{corpus, pack, vault_coherent}`` — pointers into reviewed +# in-process memory the runtime trusts as grounding. +# - ``ContemplationEvidenceRef.source_type`` is free-form because +# it points at external artifacts (benchmark report files, +# evaluator output, lab measurements) that have not been +# reviewed by the teaching pipeline. +# +# Converging them would either widen the runtime-grounding source +# enum (a real loss — losing the "this came from reviewed memory" +# guarantee) or force benchmark reports to masquerade as +# ``vault_coherent``. Both are worse than keeping the two types +# separate and documented. +# +# What IS shared: +# - ``EpistemicStatus`` (one source of truth at teaching/epistemic.py). +# - Sink plumbing (DiscoveryCandidateSink protocol) — see +# ``core/contemplation/sink.py``. +# - Append-only monthly JSONL file convention. + + @dataclass(frozen=True, slots=True) class ContemplationEvidenceRef: - """Pointer to evidence supporting a contemplation finding.""" + """Pointer to evidence supporting a contemplation finding. + + Distinct from :class:`teaching.discovery.EvidencePointer` — see + the BOUNDARY note above this class for why the two stay separate. + """ source_type: str source_id: str @@ -158,9 +189,21 @@ class ContemplationRun: } +def format_contemplation_finding_jsonl(finding: ContemplationFinding) -> str: + """Return a deterministic JSONL line for a contemplation finding. + + Mirrors :func:`teaching.discovery.format_candidate_jsonl` in style + (canonical JSON, sorted keys, no trailing newline) so both flow + through the same ``DiscoveryCandidateSink`` plumbing without + schema-aware special casing. + """ + return _canonical_json(finding.as_dict()) + + __all__ = [ "ContemplationEvidenceRef", "ContemplationFinding", "ContemplationRun", "FindingKind", + "format_contemplation_finding_jsonl", ] diff --git a/tests/test_contemplation_pipeline_convergence.py b/tests/test_contemplation_pipeline_convergence.py new file mode 100644 index 00000000..43183f84 --- /dev/null +++ b/tests/test_contemplation_pipeline_convergence.py @@ -0,0 +1,331 @@ +"""Tests for the contemplation→teaching pipeline convergence refactor. + +Pins three properties: + + 1. ContemplationFinding emission flows through the SHARED sink + protocol from teaching/discovery_sink.py — not a parallel + bespoke sink. + 2. The contradiction_detection miner survives and works under the + unified sink pattern (preserves the work from the closed #56 + while routing it through the corrected boundary). + 3. The boundary doc in core/contemplation/schema.py is intact — + EvidencePointer (teaching) and ContemplationEvidenceRef (core) + stay separate by design, not by oversight. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from core.contemplation.miners.contradiction_detection import ( + mine_contradiction_detection_report, +) +from core.contemplation.runner import ( + contemplate_contradiction_reports, + contemplate_frontier_reports, +) +from core.contemplation.schema import ( + ContemplationEvidenceRef, + ContemplationFinding, + FindingKind, + format_contemplation_finding_jsonl, +) +from core.contemplation.snapshot import ContemplationSubstrate +from teaching.discovery_sink import ( + DiscoveryBufferSink, + DiscoveryCandidateSink, + DiscoveryMonthlyFileSink, +) +from teaching.epistemic import EpistemicStatus + + +# --------------------------------------------------------------------------- +# Fixtures: real-shape sample reports for both lanes +# --------------------------------------------------------------------------- + + +def _frontier_report(path: Path) -> None: + payload = { + "benchmark_family": "frontier_compare_wave1", + "suites": [ + { + "suite": "truth_lock", + "cases": [ + { + "suite": "truth_lock", + "case_id": "known_truth", + "prompt": "What is truth?", + "passed": True, + "failures": [], + }, + { + "suite": "truth_lock", + "case_id": "unknown_relation", + "prompt": "Why does xylomorphic matter?", + "passed": False, + "failures": ["unexpected_grounding_source:vault"], + }, + ], + } + ], + } + path.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8") + + +def _contradiction_report(path: Path) -> None: + payload = { + "lane": "contradiction_detection", + "split": "public", + "version": "v1", + "cases": [ + { + "id": "CON-PUB-001", + "kind": "paired_contradiction", + "passed": True, + "flagged": True, + "contested": False, + "versor_delta": 3.28677e-07, + "versor_spike": True, + }, + { + "id": "CON-PUB-002", + "kind": "paired_contradiction", + "passed": False, + "flagged": False, + "contested": False, + "versor_delta": 0.0, + "versor_spike": False, + }, + { + "id": "CON-PUB-005", + "kind": "paired_consistent", + "passed": False, + "flagged": True, + "contested": False, + "versor_delta": 3.24412e-07, + "versor_spike": True, + }, + ], + } + path.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8") + + +# --------------------------------------------------------------------------- +# 1. Shared sink protocol — the actual integration claim +# --------------------------------------------------------------------------- + + +def test_sink_is_typed_as_discovery_candidate_sink() -> None: + """DiscoveryBufferSink must satisfy DiscoveryCandidateSink — that is + the shared protocol both pipelines now feed. + + A static structural check, but if someone widens the protocol in + teaching/discovery_sink.py without adapting the contemplation + emitter, this test surfaces the regression at the type level. + """ + sink: DiscoveryCandidateSink = DiscoveryBufferSink() + assert hasattr(sink, "emit") + sink.emit("{}") + assert sink.lines == ["{}"] + + +def test_frontier_runner_emits_findings_to_shared_sink(tmp_path: Path) -> None: + report = tmp_path / "frontier.json" + _frontier_report(report) + sink = DiscoveryBufferSink() + + run = contemplate_frontier_reports((report,), sink=sink) + + # One emission per finding, no extra noise (passing cases don't emit). + assert len(sink.lines) == 1 + assert len(run.findings) == 1 + # The emitted line round-trips to the same payload as the run blob. + emitted = json.loads(sink.lines[0]) + assert emitted == run.findings[0].as_dict() + + +def test_contradiction_runner_emits_findings_to_shared_sink(tmp_path: Path) -> None: + report = tmp_path / "contradiction.json" + _contradiction_report(report) + sink = DiscoveryBufferSink() + + run = contemplate_contradiction_reports((report,), sink=sink) + + # Two failed cases (1 missed, 1 false-flag); passing case suppressed. + assert len(sink.lines) == 2 + assert len(run.findings) == 2 + predicates = {json.loads(line)["predicate"] for line in sink.lines} + assert predicates == {"missed_contradiction", "false_contradiction_flag"} + + +def test_sink_is_optional_no_op_when_absent(tmp_path: Path) -> None: + """Existing in-memory ContemplationRun behavior is preserved when + no sink is supplied — guarantees the refactor is non-breaking.""" + report = tmp_path / "contradiction.json" + _contradiction_report(report) + + run = contemplate_contradiction_reports((report,)) # no sink kwarg + + assert len(run.findings) == 2 + assert all( + f.epistemic_status is EpistemicStatus.SPECULATIVE for f in run.findings + ) + + +def test_emission_is_jsonl_canonical(tmp_path: Path) -> None: + """The JSONL line shape must match the discovery_sink convention: + one canonical-JSON line per finding, sorted keys, no newline, + deterministic across runs.""" + report = tmp_path / "contradiction.json" + _contradiction_report(report) + sink_a = DiscoveryBufferSink() + sink_b = DiscoveryBufferSink() + + contemplate_contradiction_reports((report,), sink=sink_a) + contemplate_contradiction_reports((report,), sink=sink_b) + + assert sink_a.lines == sink_b.lines + for line in sink_a.lines: + assert "\n" not in line + # Canonical JSON: re-encoding with same options reproduces the line. + parsed = json.loads(line) + roundtrip = json.dumps( + parsed, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + assert line == roundtrip + + +def test_monthly_file_sink_persists_contemplation_findings( + tmp_path: Path, +) -> None: + """End-to-end: contemplation findings land in the SAME monthly JSONL + layout discovery candidates use — //.jsonl.""" + report = tmp_path / "contradiction.json" + _contradiction_report(report) + sink_root = tmp_path / "discovery_log" + + with DiscoveryMonthlyFileSink(sink_root) as sink: + run = contemplate_contradiction_reports((report,), sink=sink) + assert len(run.findings) == 2 + + # Find the monthly file the sink created. + written = sorted(sink_root.rglob("*.jsonl")) + assert len(written) == 1, f"expected one monthly file, got {written}" + lines = [ + line + for line in written[0].read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert len(lines) == 2 + # Layout matches discovery candidate convention /.jsonl. + rel = written[0].relative_to(sink_root) + assert len(rel.parts) == 2 # YYYY / YYYY-MM.jsonl + assert rel.parts[1].endswith(".jsonl") + + +def test_sink_emission_does_not_alter_run_blob(tmp_path: Path) -> None: + """Emitting to a sink and building the ContemplationRun blob must + produce IDENTICAL findings — the sink path is additional output, + not a parallel transformation.""" + report = tmp_path / "frontier.json" + _frontier_report(report) + + run_no_sink = contemplate_frontier_reports((report,)) + sink = DiscoveryBufferSink() + run_with_sink = contemplate_frontier_reports((report,), sink=sink) + + assert run_no_sink.run_id == run_with_sink.run_id + assert run_no_sink.as_dict() == run_with_sink.as_dict() + + +# --------------------------------------------------------------------------- +# 2. Contradiction miner — preserved from closed #56, now under the +# unified sink contract +# --------------------------------------------------------------------------- + + +def test_contradiction_miner_predicate_split(tmp_path: Path) -> None: + report = tmp_path / "contradiction.json" + _contradiction_report(report) + substrate = ContemplationSubstrate.from_report_paths((report,)) + + findings = mine_contradiction_detection_report( + report, substrate_hash=substrate.substrate_hash + ) + by_predicate = {f.predicate: f for f in findings} + assert set(by_predicate.keys()) == { + "missed_contradiction", + "false_contradiction_flag", + } + # Repair-action asymmetry: tightening (missed) vs loosening (false flag). + assert "tighten" in by_predicate["missed_contradiction"].proposed_action + assert "loosen" in by_predicate["false_contradiction_flag"].proposed_action + # All findings emit FindingKind.CONTRADICTION (not BENCHMARK_CASE) — + # the lane is semantically targeted. + for finding in findings: + assert finding.kind is FindingKind.CONTRADICTION + + +def test_contradiction_config_hash_differs_from_frontier(tmp_path: Path) -> None: + """Lane choice must remain load-bearing in config_hash so replay + can distinguish which lane was contemplated. Regression guard + against a future refactor that accidentally unifies the runners.""" + report = tmp_path / "shared.json" + _contradiction_report(report) + a = contemplate_contradiction_reports((report,)) + b = contemplate_frontier_reports((report,)) + assert a.config_hash != b.config_hash + + +# --------------------------------------------------------------------------- +# 3. Boundary doc — the deliberate non-merger of evidence types +# --------------------------------------------------------------------------- + + +def test_boundary_doc_present_in_schema_module() -> None: + """The BOUNDARY note in core/contemplation/schema.py is the + durable record of why EvidencePointer and ContemplationEvidenceRef + stay separate. Pinning its presence prevents a future contributor + from silently deleting the boundary rationale.""" + schema_text = Path("core/contemplation/schema.py").read_text(encoding="utf-8") + assert "BOUNDARY" in schema_text + assert "EvidencePointer" in schema_text + assert "vault_coherent" in schema_text + + +def test_contemplation_evidence_ref_rejects_empty_fields() -> None: + """The existing schema invariants must survive the refactor.""" + with pytest.raises(ValueError, match="source_type"): + ContemplationEvidenceRef(source_type="", source_id="x", pointer="y") + with pytest.raises(ValueError, match="source_id"): + ContemplationEvidenceRef(source_type="x", source_id="", pointer="y") + with pytest.raises(ValueError, match="pointer"): + ContemplationEvidenceRef(source_type="x", source_id="y", pointer="") + + +def test_format_contemplation_finding_jsonl_is_deterministic() -> None: + finding = ContemplationFinding( + kind=FindingKind.CONTRADICTION, + subject="lane/case_001", + predicate="missed_contradiction", + object=None, + evidence_refs=( + ContemplationEvidenceRef( + source_type="t", source_id="i", pointer="p", summary="s" + ), + ), + proposed_action="review", + substrate_hash="abc123", + ) + a = format_contemplation_finding_jsonl(finding) + b = format_contemplation_finding_jsonl(finding) + assert a == b + # Round-trips through json.loads/dumps with same options. + parsed = json.loads(a) + assert ( + json.dumps(parsed, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + == a + )