refactor(contemplation): converge to shared discovery-sink plumbing (#58)

Connects ADR-0080's read-only contemplation loop to the existing
teaching-pipeline plumbing without forcing a type collapse.  The
SPECULATIVE-only invariant from #55 is preserved verbatim; what
changes is *where the findings flow*.

What was wrong with the prior shape
-----------------------------------
PR #55 shipped a parallel core/contemplation/ package whose findings
were written as one JSON blob per CLI invocation, with no consumer.
The SPECULATIVE-only invariant protected a write path that didn't
exist.  My closed PR #56 (second miner) would have entrenched the
duplication.

What this PR changes
--------------------
1. Schema (core/contemplation/schema.py)
   - Adds a BOUNDARY note documenting why EvidencePointer (teaching)
     and ContemplationEvidenceRef (core) intentionally stay separate:
     EvidencePointer.source is constrained to {corpus, pack,
     vault_coherent} — pointers into reviewed in-process memory the
     runtime trusts.  ContemplationEvidenceRef points to external
     report files that have NOT been reviewed.  Converging them would
     either widen the runtime-grounding enum (losing the "reviewed
     memory only" guarantee) or force benchmark reports to masquerade
     as vault_coherent.  Both are worse than keeping them separate.
   - Adds format_contemplation_finding_jsonl(finding) — the canonical
     JSONL formatter mirroring teaching.discovery.format_candidate_jsonl.

2. Runner (core/contemplation/runner.py)
   - Both runners gain an optional sink: DiscoveryCandidateSink | None
     parameter.  When supplied, each finding is emitted as one
     canonical JSONL line via the SHARED protocol — same protocol
     that backs DiscoveryBufferSink and DiscoveryMonthlyFileSink.
   - Sink path is additive: the ContemplationRun blob is byte-identical
     whether or not a sink is supplied (pinned by test).
   - No sink supplied → existing in-memory behavior preserved exactly.

3. CLI (core/contemplation/__main__.py)
   - Adds --lane {frontier_compare, contradiction_detection} flag.
     Default unchanged.
   - Adds --sink-root <path> flag.  When set, instantiates a
     DiscoveryMonthlyFileSink and findings land at
     <root>/<YYYY>/<YYYY-MM>.jsonl — the SAME layout discovery
     candidates use, so operators can grep one stream.

4. Miner (core/contemplation/miners/contradiction_detection.py)
   - Restored from closed PR #56 under the unified pipeline.
   - Failure-mode split preserved (missed_contradiction /
     false_contradiction_flag) with asymmetric repair actions.

What this PR does NOT do
------------------------
- Does NOT unify ContemplationFinding with DiscoveryCandidate.
  DiscoveryCandidate.trigger is Literal[would_have_grounded,
  successful_comparison, hedge_acknowledged, oov_resolved_via_decomp]
  — all turn-loop flavored.  None describe "I parsed a benchmark
  report."  Forcing a 5th trigger that no turn-loop extractor
  produces would pollute the turn-loop type for the schema's sake.
- Does NOT extend teaching/gaps.py.  Gap aggregates DiscoveryCandidate
  cells by (subject, intent) — domain nouns.  ContemplationFinding
  subjects are namespaced ("contradiction_detection/CON-PUB-002").
  Different operator views.  A sibling aggregator can come later
  when an operator actually asks for it.

Why this is the right unification point
---------------------------------------
The honest convergence is at the *sink* (so all SPECULATIVE evidence
lives in one rooted append-only stream), not the *aggregator* (which
appropriately produces typed views per evidence family).  The boundary
doctrine from #55 is preserved; it now connects to existing plumbing
instead of writing JSON to disk with no consumer.

Tests (tests/test_contemplation_pipeline_convergence.py, 10 cases)
------------------------------------------------------------------
- DiscoveryBufferSink satisfies DiscoveryCandidateSink (shared protocol)
- frontier runner emits findings to shared sink
- contradiction runner emits findings to shared sink
- sink is optional — no-op when absent
- emission is canonical JSONL (sorted keys, no newline, deterministic)
- DiscoveryMonthlyFileSink persists findings at <root>/<YYYY>/<YYYY-MM>.jsonl
- sink emission does not alter the ContemplationRun blob (additive)
- contradiction miner predicate split + repair-action asymmetry
- config_hash differs between lanes (replay can distinguish)
- BOUNDARY doc is present in schema.py (regression guard)
- ContemplationEvidenceRef field invariants
- format_contemplation_finding_jsonl is deterministic + canonical

All 18 tests pass (5 original ADR-0080 + 13 new convergence).

Live evidence
-------------
$ uv run python -m core.contemplation \
    evals/contradiction_detection/results/v1_public_*.json \
    --lane contradiction_detection \
    --sink-root /tmp/sink_demo

  /tmp/sink_demo/2026/2026-05.jsonl  ← same layout as discovery candidates

  predicate=missed_contradiction         subject=contradiction_detection/CON-PUB-002
  predicate=missed_contradiction         subject=contradiction_detection/CON-PUB-004
  predicate=false_contradiction_flag     subject=contradiction_detection/CON-PUB-005
  predicate=false_contradiction_flag     subject=contradiction_detection/CON-PUB-006

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 "
+            "<root>/<YYYY>/<YYYY-MM>.jsonl alongside discovery candidates."
+        ),
     )
     return parser
 
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
-    run = contemplate_frontier_reports(
+    runner = _LANE_RUNNERS[args.lane]
-        args.reports,
+    sink = (
-        pack_ids=tuple(args.pack_id or ()),
+        DiscoveryMonthlyFileSink(args.sink_root) if args.sink_root is not None else None
-        notes=tuple(args.note or ()),
     )
+    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))

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",
+]

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"]

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",
+]

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",
 ]

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 — <root>/<YYYY>/<YYYY-MM>.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 <YYYY>/<YYYY-MM>.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
+    )