diff --git a/docs/cli-reference.md b/docs/cli-reference.md
index 9c9875113f..d3a66c41ef 100644
--- a/docs/cli-reference.md
+++ b/docs/cli-reference.md
@@ -205,6 +205,7 @@ Commands:
   Other commands:
     agent        Install executable agent guidance.
     annotations  Import typed annotation batches.
+    compare      Blind pairwise comparative judgment and calibration.
 ```
 
 ## Analyze Verb
diff --git a/polylogue/api/archive.py b/polylogue/api/archive.py
index 9c113755a5..e5ee55fc31 100644
--- a/polylogue/api/archive.py
+++ b/polylogue/api/archive.py
@@ -104,6 +104,7 @@
     from polylogue.insights.audit import InsightRigorAuditQuery, InsightRigorAuditReport
     from polylogue.insights.export_bundles import InsightExportBundleRequest, InsightExportBundleResult
     from polylogue.insights.hermes_integration_health import HermesIntegrationHealth
+    from polylogue.insights.judgment.types import ComparativeJudgment
     from polylogue.insights.pathology import PathologyReport
     from polylogue.insights.portfolio import PortfolioBundle
     from polylogue.insights.postmortem import PostmortemBundle
@@ -1816,6 +1817,58 @@ def _archive_judge_assertion_candidates(
         raise RuntimeError(f"failed to judge assertion candidates: {exc}") from exc
 
 
+def _archive_record_comparative_judgment(
+    config: Config,
+    judgment: Any,
+    *,
+    author_kind: str,
+) -> Any:
+    """Write one comparative judgment (rxdo.9.11/9.6/9.7/9.12) as an assertion row.
+
+    Mirrors :func:`_archive_judge_assertion_candidates`'s connection
+    lifecycle. This is the first production caller of
+    :func:`~polylogue.storage.sqlite.archive_tiers.user_write.upsert_comparative_judgment_assertion`
+    -- the storage/read functions were fully built and tested but never
+    invoked outside ``tests/unit/storage/`` before the ``judge compare`` /
+    ``judge calibration`` CLI commands.
+    """
+    from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+    from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+    from polylogue.storage.sqlite.archive_tiers.user_write import upsert_comparative_judgment_assertion
+
+    user_db = _active_archive_root(config) / "user.db"
+    initialize_archive_database(user_db, ArchiveTier.USER)
+    try:
+        conn = open_connection(user_db)
+        conn.row_factory = sqlite3.Row
+        try:
+            envelope = upsert_comparative_judgment_assertion(conn, judgment, author_kind=author_kind)
+            conn.commit()
+            return envelope
+        finally:
+            conn.close()
+    except sqlite3.Error as exc:
+        raise RuntimeError(f"failed to record comparative judgment: {exc}") from exc
+
+
+def _archive_list_comparative_judgments(config: Config) -> Any:
+    """Read back every live comparative-judgment assertion row."""
+
+    from polylogue.storage.sqlite.archive_tiers.user_write import list_comparative_judgments
+
+    user_db = _active_archive_root(config) / "user.db"
+    if not user_db.exists():
+        return []
+    try:
+        conn = open_readonly_connection(user_db)
+        try:
+            return list_comparative_judgments(conn)
+        finally:
+            conn.close()
+    except sqlite3.Error as exc:
+        raise RuntimeError(f"failed to list comparative judgments: {exc}") from exc
+
+
 def _archive_count_table_rows(conn: Any, table_name: str) -> int | None:
     row = conn.execute(
         "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'view') AND name = ? LIMIT 1",
@@ -3022,6 +3075,30 @@ async def judge_assertion_candidates(
         result = _archive_judge_assertion_candidates(self.config, items=items)
         return AssertionBulkJudgmentPayload.from_envelope(cast("ArchiveAssertionBulkJudgmentEnvelope", result))
 
+    async def record_comparative_judgment(
+        self,
+        judgment: ComparativeJudgment,
+        *,
+        author_kind: str = "user",
+    ) -> ArchiveAssertionEnvelope:
+        """Persist one blind pairwise/n-wise comparative judgment (rxdo.9.6/.9.7/.9.11/.9.12).
+
+        ``author_kind`` follows the existing promotion gate: a non-``"user"``
+        author (an agent judge) is coerced to a non-injected ``CANDIDATE``
+        row regardless of the caller's request, per the recursive-safety
+        spine. Reuses the storage layer's fully-tested write chokepoint
+        (:func:`~polylogue.storage.sqlite.archive_tiers.user_write.upsert_comparative_judgment_assertion`),
+        which previously had no production caller.
+        """
+        return cast(
+            "ArchiveAssertionEnvelope",
+            _archive_record_comparative_judgment(self.config, judgment, author_kind=author_kind),
+        )
+
+    async def list_comparative_judgments(self) -> list[ComparativeJudgment]:
+        """Read back every live comparative-judgment assertion row."""
+        return cast("list[ComparativeJudgment]", _archive_list_comparative_judgments(self.config))
+
     async def join_typed_annotations(
         self,
         *,
@@ -3938,6 +4015,7 @@ def _resolve_finding_object_ref(
         with closing(open_readonly_connection(user_db)) as conn:
             conn.row_factory = sqlite3.Row
             provenance = compute_finding_provenance(conn, object_ref.object_id)
+            controls_document = self._finding_controls_document(conn, object_ref.object_id)
         if provenance is None:
             return cast(
                 PublicRefResolutionPayload,
@@ -3977,13 +4055,21 @@ def _resolve_finding_object_ref(
                 if ref_value
             )
         )
+        payload_document = model_json_document(payload)
+        if controls_document is not None:
+            payload_document["controls"] = controls_document["controls"]
+            payload_document["rank_tier"] = controls_document["rank_tier"]
+            payload_document["downgraded"] = controls_document["downgraded"]
+            if controls_document["downgraded"]:
+                caveats = (*caveats, "claim downgraded: at least one bound negative control failed")
+
         return PublicRefResolutionPayload(
             ref=ref,
             normalized_ref=normalized_ref,
             kind="finding",
             resolved=True,
             payload_kind="finding-provenance",
-            payload=model_json_document(payload),
+            payload=payload_document,
             title=provenance.claim_key or provenance.finding_kind or "finding",
             summary=f"{provenance.finding_kind or 'finding'} ({provenance.status})",
             object_refs=object_refs,
@@ -3992,6 +4078,54 @@ def _resolve_finding_object_ref(
             actions=(_resolution_action("list target evidence", f"polylogue find {provenance.target_ref} then read"),),
         )
 
+    @staticmethod
+    def _finding_controls_document(conn: Any, assertion_id: str) -> dict[str, Any] | None:
+        """Render claim-vs-control together when the finding declared controls (rxdo.9.7).
+
+        Reuses :class:`~polylogue.insights.judgment.controls.ClaimWithControls`
+        (mutation-tested, previously constructed only by its own unit tests)
+        rather than re-deriving the downgrade/rank-tier logic here.
+        """
+        from polylogue.insights.judgment.controls import ClaimWithControls, ControlOutcome, NegativeControl
+        from polylogue.storage.sqlite.archive_tiers.user_write import read_assertion_envelope
+
+        envelope = read_assertion_envelope(conn, assertion_id)
+        if envelope is None or not isinstance(envelope.value, dict):
+            return None
+        raw_controls = envelope.value.get("controls")
+        if not isinstance(raw_controls, list) or not raw_controls:
+            return None
+        outcomes = tuple(
+            ControlOutcome(
+                control=NegativeControl(
+                    control_kind=cast(Any, raw["control_kind"]),
+                    query_ref=cast(str, raw["query_ref"]),
+                    result_ref=cast(str, raw["result_ref"]),
+                    matching_variables=tuple(cast("Sequence[str]", raw.get("matching_variables", ()))),
+                    expected_null=cast(str, raw["expected_null"]),
+                    confounds_checked=tuple(cast("Sequence[str]", raw.get("confounds_checked", ()))),
+                ),
+                observed_null_held=bool(raw["observed_null_held"]),
+            )
+            for raw in raw_controls
+            if isinstance(raw, dict)
+        )
+        claim = ClaimWithControls(claim_ref=f"assertion:{assertion_id}", controls=outcomes)
+        return {
+            "controls": [
+                {
+                    "control_kind": outcome.control.control_kind,
+                    "query_ref": outcome.control.query_ref,
+                    "result_ref": outcome.control.result_ref,
+                    "expected_null": outcome.control.expected_null,
+                    "observed_null_held": outcome.observed_null_held,
+                }
+                for outcome in claim.controls
+            ],
+            "rank_tier": claim.rank_tier,
+            "downgraded": claim.downgraded,
+        }
+
     def _resolve_annotation_batch_object_ref(
         self,
         archive: Any,
diff --git a/polylogue/cli/click_command_registration.py b/polylogue/cli/click_command_registration.py
index 5323b0df97..d16c838b4e 100644
--- a/polylogue/cli/click_command_registration.py
+++ b/polylogue/cli/click_command_registration.py
@@ -91,6 +91,7 @@ def list_commands(self, ctx: click.Context) -> list[str]:
     "auth": "Authenticate optional external services.",
     "backup": "Create a timestamped durability-tier backup.",
     "check": "Run archive health checks and repairs.",
+    "compare": "Blind pairwise comparative judgment and calibration.",
     "completions": "Emit shell completion setup for polylogue.",
     "config": "Show resolved Polylogue configuration with...",
     "dashboard": "Launch the terminal dashboard TUI.",
@@ -165,6 +166,7 @@ def _L(name: str) -> _LazyCommand:  # noqa: N802
     _L("agent"),
     _L("agents"),
     _L("annotations"),
+    _L("compare"),
     _L("config"),
     _L("dashboard"),
     _L("demo"),
diff --git a/polylogue/cli/commands/compare.py b/polylogue/cli/commands/compare.py
new file mode 100644
index 0000000000..25a0708790
--- /dev/null
+++ b/polylogue/cli/commands/compare.py
@@ -0,0 +1,207 @@
+"""Blind pairwise comparative judgment (rxdo.9.6/.9.7/.9.11/.9.12).
+
+Wires the previously-unwired judgment mechanisms
+(``polylogue/insights/judgment/``) into a real command surface: blinding
+(rxdo.9.6, ``blinding.py``) masks provenance before verdict; the recorded
+verdict persists through the fully-built but previously-uncalled storage
+chokepoint (``upsert_comparative_judgment_assertion``, rxdo.9.11); and
+calibration (rxdo.9.12, ``calibration.py``) reports per-judge agreement with
+a designated gold actor over the recorded judgments.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import asdict
+
+import click
+
+from polylogue.api.sync.bridge import run_coroutine_sync
+from polylogue.cli.shared.types import AppEnv
+from polylogue.core.enums import ComparativeVerdict
+from polylogue.core.refs import ActorRef, ExecutionContextRef
+from polylogue.insights.judgment.blinding import assert_no_leak, blind_items, reveal
+from polylogue.insights.judgment.calibration import compute_calibration
+from polylogue.insights.judgment.comparative import build_comparative_judgment
+from polylogue.insights.judgment.types import JudgeIdentity
+
+_VERDICT_CHOICES = tuple(verdict.value for verdict in ComparativeVerdict)
+
+
+def _parse_fields(pairs: tuple[str, ...]) -> dict[str, object]:
+    fields: dict[str, object] = {}
+    for pair in pairs:
+        key, sep, value = pair.partition("=")
+        if not sep:
+            raise click.UsageError(f"--left-field/--right-field must be key=value, got {pair!r}")
+        fields[key] = value
+    return fields
+
+
+def _judge_identity(actor_ref: str, exec_context_id: str) -> JudgeIdentity:
+    return JudgeIdentity(
+        actor=ActorRef.parse(actor_ref), execution_context=ExecutionContextRef.from_legacy_id(exec_context_id)
+    )
+
+
+@click.command("compare")
+@click.option("--left", "left_ref", default=None, help="Left item ref.")
+@click.option("--right", "right_ref", default=None, help="Right item ref.")
+@click.option("--left-field", "left_fields", multiple=True, help="Provenance field on the left item, key=value.")
+@click.option("--right-field", "right_fields", multiple=True, help="Provenance field on the right item, key=value.")
+@click.option("--dimension", default=None, help="Comparison dimension (e.g. quality, correctness).")
+@click.option("--rubric", "rubric_id", default=None, help="Rubric identifier the blinded receipt binds to.")
+@click.option("--rubric-version", type=int, default=1, show_default=True)
+@click.option(
+    "--verdict",
+    type=click.Choice(_VERDICT_CHOICES),
+    default=None,
+    help="Record this verdict (omit to only print the blinded pair without recording).",
+)
+@click.option("--rationale", default=None, help="Optional judge rationale (stored only if --rationale-visible).")
+@click.option("--rationale-visible", is_flag=True, help="Store rationale visibly rather than redacted.")
+@click.option("--evidence-ref", "evidence_refs", multiple=True, help="Additional evidence ref for the judgment.")
+@click.option("--actor-ref", default="user:local", show_default=True)
+@click.option("--exec-context-id", default="cli:compare", show_default=True)
+@click.option("--calibration", "calibration_mode", is_flag=True, help="Report calibration instead of comparing.")
+@click.option("--gold-actor", default=None, help="Actor ref treated as gold for --calibration.")
+@click.option("--json", "output_format", flag_value="json", default=None, help="Shortcut for --format json.")
+@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default=None)
+@click.pass_obj
+def compare_command(
+    env: AppEnv,
+    left_ref: str | None,
+    right_ref: str | None,
+    left_fields: tuple[str, ...],
+    right_fields: tuple[str, ...],
+    dimension: str | None,
+    rubric_id: str | None,
+    rubric_version: int,
+    verdict: str | None,
+    rationale: str | None,
+    rationale_visible: bool,
+    evidence_refs: tuple[str, ...],
+    actor_ref: str,
+    exec_context_id: str,
+    calibration_mode: bool,
+    gold_actor: str | None,
+    output_format: str | None,
+) -> None:
+    """Blind pairwise comparative judgment, or a calibration report over recorded judgments."""
+
+    output_format = output_format or "text"
+
+    if calibration_mode:
+        if gold_actor is None:
+            raise click.UsageError("--calibration requires --gold-actor")
+        judgments = run_coroutine_sync(env.polylogue.list_comparative_judgments())
+        gold = [j for j in judgments if j.judge.actor_ref == gold_actor]
+        candidates = [j for j in judgments if j.judge.actor_ref != gold_actor]
+        reports = compute_calibration(candidates, gold)
+        if output_format == "json":
+            click.echo(
+                json.dumps(
+                    [
+                        {
+                            "actor_ref": key.actor_ref,
+                            "execution_context_id": key.execution_context_id,
+                            "dimension": key.dimension,
+                            "n_gold_overlap": report.n_gold_overlap,
+                            "agreement_rate": report.agreement_rate,
+                            "tie_rate": report.tie_rate,
+                            "incomparable_rate": report.incomparable_rate,
+                            "abstain_rate": report.abstain_rate,
+                            "insufficient_evidence_rate": report.insufficient_evidence_rate,
+                            "n_total_verdicts": report.n_total_verdicts,
+                        }
+                        for key, report in reports.items()
+                    ],
+                    indent=2,
+                )
+            )
+            return
+        if not reports:
+            click.echo("No comparative judgments to calibrate.")
+            return
+        for key, report in reports.items():
+            agreement = "unknown (no gold overlap)" if report.agreement_rate is None else f"{report.agreement_rate:.2%}"
+            click.echo(
+                f"{key.actor_ref} @ {key.execution_context_id} / {key.dimension}: "
+                f"agreement={agreement} n={report.n_total_verdicts} gold_overlap={report.n_gold_overlap}"
+            )
+        return
+
+    if left_ref is None or right_ref is None or dimension is None or rubric_id is None:
+        raise click.UsageError("--left, --right, --dimension, and --rubric are required (or use --calibration)")
+
+    left_record: dict[str, object] = {"ref": left_ref, **_parse_fields(left_fields)}
+    right_record: dict[str, object] = {"ref": right_ref, **_parse_fields(right_fields)}
+    sealed_at_ms = int(time.time() * 1000)
+    blinded, receipt = blind_items(
+        [left_record, right_record],
+        order=[0, 1],
+        rubric_ref=rubric_id,
+        sealed_at_ms=sealed_at_ms,
+    )
+    assert_no_leak(blinded)
+
+    if verdict is None:
+        payload = {
+            "items": [asdict(item) for item in blinded],
+            "receipt": asdict(receipt),
+        }
+        click.echo(json.dumps(payload, indent=2) if output_format == "json" else _render_blinded_text(blinded, receipt))
+        return
+
+    judge = _judge_identity(actor_ref, exec_context_id)
+    verdict_enum = ComparativeVerdict.from_string(verdict)
+    judgment = build_comparative_judgment(
+        items=[left_ref, right_ref],
+        dimension=dimension,
+        verdict=verdict_enum,
+        judge=judge,
+        blinded=True,
+        rubric_id=rubric_id,
+        rubric_version=rubric_version,
+        decided_at_ms=int(time.time() * 1000),
+        evidence_refs=evidence_refs,
+        rationale=rationale,
+        rationale_visible=rationale_visible,
+    )
+    envelope = run_coroutine_sync(env.polylogue.record_comparative_judgment(judgment, author_kind="user"))
+    revealed_receipt = reveal(receipt, revealed_at_ms=int(time.time() * 1000), verdict_recorded=True)
+
+    if output_format == "json":
+        click.echo(
+            json.dumps(
+                {
+                    "judgment_id": judgment.judgment_id,
+                    "assertion_id": envelope.assertion_id,
+                    "status": envelope.status.value,
+                    "verdict": verdict_enum.value,
+                    "revealed": {"left": left_record, "right": right_record},
+                    "receipt": asdict(revealed_receipt),
+                },
+                indent=2,
+            )
+        )
+        return
+    click.echo(f"Recorded {judgment.judgment_id} ({verdict_enum.value}) as assertion {envelope.assertion_id}.")
+    click.echo(f"  revealed left:  {left_record}")
+    click.echo(f"  revealed right: {right_record}")
+
+
+def _render_blinded_text(blinded: object, receipt: object) -> str:
+    from polylogue.insights.judgment.blinding import BlindedItem, BlindingReceipt
+
+    assert isinstance(blinded, tuple)
+    assert isinstance(receipt, BlindingReceipt)
+    lines = [f"rubric: {receipt.rubric_ref}  masked_fields: {list(receipt.masked_fields)}"]
+    for item in blinded:
+        assert isinstance(item, BlindedItem)
+        lines.append(f"  [{item.display_position}] {dict(item.visible_fields)}")
+    return "\n".join(lines)
+
+
+__all__ = ["compare_command"]
diff --git a/polylogue/insights/measurement/registered_metrics.py b/polylogue/insights/measurement/registered_metrics.py
new file mode 100644
index 0000000000..fe6b5db527
--- /dev/null
+++ b/polylogue/insights/measurement/registered_metrics.py
@@ -0,0 +1,57 @@
+"""The process-wide default metric registry (rxdo.9.1 identity/schema layer).
+
+:class:`~polylogue.insights.measurement.metric.MetricDefinition` and
+:class:`~polylogue.insights.measurement.metric.MetricRegistry` (PR #2888,
+merged) had zero production callers -- the corrective AC's second consumer
+path ("one hash resolves through both query/analysis and statistical-
+registry paths") depends on ``polylogue-9l5.7``'s statistics registry, which
+remains unstarted. Building that whole registry/composition epic is out of
+scope here (see ``polylogue-rxdo.9`` epic-expansion guard).
+
+This module is the bounded, honest slice available without it: a real,
+process-wide registry populated with one concrete metric definition for an
+existing, already-computed construct (session cost, ``cost/pricing.py``/
+``cost/outlook.py``), reachable through the MCP ``get`` tool
+(``get(ref="metric:session_cost_usd")``, see ``polylogue/mcp/
+server_cutover.py``). This proves the identity/registry machinery resolves
+through a real production surface -- it does NOT execute the metric (no
+composition/aggregation engine exists yet; that is 9l5.7's job) or attach a
+``metric_ref`` to computed values anywhere. Both remain open scope.
+"""
+
+from __future__ import annotations
+
+from polylogue.insights.measurement.metric import MetricDefinition, MetricRegistry
+
+#: Session-level USD cost: provider-reported totals where available,
+#: catalog-priced (LiteLLM) estimates otherwise -- mirrors the mixed-basis
+#: reality already documented in ``docs/cost-model.md`` and computed by
+#: ``polylogue/archive/semantic/pricing.py:estimate_session_cost`` /
+#: ``polylogue/cost/outlook.py``. Declared ``mixed-declared`` rather than
+#: ``single-authority`` because the two lanes (provider-reported vs.
+#: catalog-estimated) are intentionally blended, per the 9l5.7 bead's own
+#: denominator-hazards checklist ("outcome_conditioned_cost must never
+#: silently mix provider-reported with catalog estimates" -- this metric
+#: names the mixing explicitly instead).
+SESSION_COST_USD_METRIC = MetricDefinition(
+    construct="total estimated USD cost for one session",
+    unit="usd",
+    unit_source="session_costs",
+    aggregation="sum",
+    grain="logical",
+    required_enumeration="exact",
+    measurement_authority=("provider-reported", "catalog-estimated"),
+    provenance_mixing="mixed-declared",
+    output_schema="usd:float",
+)
+
+#: Process-wide default registry. A module-level singleton is the correct
+#: shape for an in-process content-addressed identity registry (mirrors
+#: ``polylogue.insights.registry.INSIGHT_REGISTRY``) -- registration is
+#: idempotent by content hash, so re-importing this module never double
+#: -registers or drifts.
+DEFAULT_METRIC_REGISTRY = MetricRegistry()
+DEFAULT_METRIC_REGISTRY.register(SESSION_COST_USD_METRIC, name="session_cost_usd")
+
+
+__all__ = ["DEFAULT_METRIC_REGISTRY", "SESSION_COST_USD_METRIC"]
diff --git a/polylogue/mcp/server_cutover.py b/polylogue/mcp/server_cutover.py
index da52a625f0..e9526f4b29 100644
--- a/polylogue/mcp/server_cutover.py
+++ b/polylogue/mcp/server_cutover.py
@@ -50,6 +50,150 @@ def _object_ref(ref: str) -> str:
     return f"{prefix}:{object_id}"
 
 
+async def _resolve_reference_query_pipeline(
+    hooks: ServerCallbacks, expression: str, *, limit: int | None
+) -> str | None:
+    """Resolve a ``from query:<hash>|result-set:<id>|query-run:<id>|cohort:<id>`` pipeline.
+
+    Returns ``None`` when ``expression`` is not a reference pipeline (the
+    caller falls through to the ordinary DSL path). Otherwise resolves the
+    root operand through the real planner seam (``DurableRefResolver`` +
+    ``ArchiveCanonicalPlanEvaluator``, both production implementations as of
+    PR #2899) and returns its member refs with lineage -- rxdo.6's "next
+    concrete slice": wire ONE command surface to call the reference-aware
+    planner instead of hard-erroring. Stage composition after the root
+    operand (``| group by ... | count``) is not implemented yet; a pipeline
+    with stages returns a typed ``not_implemented`` error naming the gap
+    rather than silently ignoring the stages or crashing.
+    """
+    import sqlite3
+    from contextlib import closing
+
+    from polylogue.archive.query.evaluator import RetainedRelationUnavailableError
+    from polylogue.archive.query.expression import (
+        ExpressionCompileError,
+        RefOperandCycleError,
+        parse_reference_query_pipeline,
+        resolve_ref_operand,
+    )
+    from polylogue.archive.query.production_evaluator import (
+        ArchiveCanonicalPlanEvaluator,
+        LegacyQueryDefinitionNotExecutableError,
+        UnsupportedEvaluationGrainError,
+    )
+    from polylogue.mcp.archive_support import mcp_archive_root
+
+    pipeline = parse_reference_query_pipeline(expression)
+    if pipeline is None:
+        return None
+    if pipeline.stages:
+        return hooks.error_json(
+            "reference-pipeline stage composition (e.g. `| group by ...`, `| count`) is not "
+            "implemented yet; only the bare `from <ref>` operand resolves. See polylogue-rxdo.6.",
+            code="not_implemented",
+            tool="query",
+        )
+
+    archive_root = mcp_archive_root(hooks.get_config())
+    user_db = archive_root / "user.db"
+    index_db = archive_root / "index.db"
+    if not user_db.exists() or not index_db.exists():
+        return hooks.error_json("archive is not initialized", code="not_found", tool="query")
+
+    evaluator = ArchiveCanonicalPlanEvaluator(index_db, surface="mcp")
+    try:
+        with closing(sqlite3.connect(f"file:{user_db}?mode=ro", uri=True, timeout=5.0)) as conn:
+            from polylogue.archive.query.evaluator import DurableRefResolver
+
+            resolver = DurableRefResolver(conn, evaluator)
+            resolved = resolve_ref_operand(pipeline.operand, resolver)
+    except KeyError:
+        return hooks.error_json(
+            f"reference not found: {pipeline.operand.reference.format()}", code="not_found", tool="query"
+        )
+    except (
+        RetainedRelationUnavailableError,
+        RefOperandCycleError,
+        ExpressionCompileError,
+        LegacyQueryDefinitionNotExecutableError,
+        UnsupportedEvaluationGrainError,
+        NotImplementedError,
+    ) as exc:
+        return hooks.error_json(str(exc), code="invalid_argument", tool="query")
+
+    member_refs = resolved.member_refs
+    truncated = False
+    if limit is not None and limit >= 0 and len(member_refs) > limit:
+        member_refs = member_refs[:limit]
+        truncated = True
+    return hooks.json_payload(
+        MCPRootPayload(
+            root={
+                "source": pipeline.operand.reference.format(),
+                "grain": resolved.grain,
+                "lineage": [ref.format() for ref in resolved.lineage],
+                "member_count": len(resolved.member_refs),
+                "members": member_refs,
+                "truncated": truncated,
+            }
+        )
+    )
+
+
+def _metric_definition_payload(hooks: ServerCallbacks, metric_id: str) -> str:
+    """Resolve a ``metric:<hash-or-registered-name>`` ref (rxdo.9.1 identity layer).
+
+    Looks up the friendly name first (the common case for a hand-typed
+    ref), falling back to a direct content-hash lookup for
+    ``metric:<hash>`` refs copied from another surface's output.
+    """
+    from polylogue.insights.measurement.registered_metrics import DEFAULT_METRIC_REGISTRY
+
+    definition = DEFAULT_METRIC_REGISTRY.resolve(metric_id) or DEFAULT_METRIC_REGISTRY.get(f"metric:{metric_id}")
+    if definition is None:
+        return hooks.error_json(f"metric not found in the default registry: metric:{metric_id}", code="not_found")
+    return hooks.json_payload(
+        MCPRootPayload(root={"ref": definition.ref, "definition": definition.canonical_payload()}),
+        exclude_none=True,
+    )
+
+
+async def _cost_outlook_payload(hooks: ServerCallbacks, *, plan_name: str, method: str | None) -> str:
+    """Project the current billing cycle for ``plan_name`` (``get`` tool, ``cost-outlook:`` refs).
+
+    Mirrors the CLI ``analyze --cost-outlook`` call shape
+    (``polylogue/cli/query_verbs.py``) so both surfaces share one
+    ``Polylogue.cost_outlook`` production route rather than redefining the
+    projection or its "no cycle window" degradation here.
+    """
+    from polylogue.cost.outlook import ProjectionMethod
+    from polylogue.cost.plans import PlanLookupError
+    from polylogue.insights.projection_contracts import cost_outlook_availability
+
+    if not plan_name.strip():
+        return hooks.error_json("cost-outlook ref requires a plan name", code="invalid_argument", tool="get")
+    try:
+        projection_method = ProjectionMethod(method) if method else ProjectionMethod.linear
+    except ValueError:
+        valid = ", ".join(item.value for item in ProjectionMethod)
+        return hooks.error_json(
+            f"unsupported cost-outlook projection {method!r}; expected one of: {valid}",
+            code="invalid_argument",
+            tool="get",
+        )
+    try:
+        outlook = await hooks.get_polylogue().cost_outlook(plan_name, method=projection_method)
+    except PlanLookupError as exc:
+        return hooks.error_json(str(exc), code="invalid_argument", tool="get")
+    if outlook is None:
+        availability = cost_outlook_availability(plan_name, ready=False, elapsed_s=0.0)
+        return hooks.json_payload(
+            MCPRootPayload(root={"outlook": None, "availability": availability.model_dump(mode="json")}),
+            exclude_none=True,
+        )
+    return hooks.json_payload(outlook, exclude_none=True)
+
+
 async def _query_sessions(
     hooks: ServerCallbacks,
     *,
@@ -571,6 +715,11 @@ async def query(
         """
 
         async def run() -> str:
+            if expression is not None:
+                reference_result = await _resolve_reference_query_pipeline(hooks, expression, limit=limit)
+                if reference_result is not None:
+                    return reference_result
+
             if projection == "sessions":
                 if continuation is not None:
                     return hooks.error_json(
@@ -681,11 +830,30 @@ async def get(ref: str, projection: str | None = None) -> str:
         ``turn_context`` policy facts, Claude Code sidecar events, Hermes
         tool-availability spans, and similar provider evidence that rides the
         timeline rather than a dialogue message.
+
+        ``ref="cost-outlook:<plan_name>"`` projects the current billing cycle
+        for a configured subscription plan (the standalone ``cost_outlook``
+        MCP tool retired by the six-tool cutover, #3095/polylogue-t46.8, has
+        no replacement otherwise -- see polylogue-hg97). ``projection``
+        selects the projection method (``linear`` default, ``trailing-7d-mean``,
+        or ``eom-naive``).
+
+        ``ref="metric:<hash-or-registered-name>"`` resolves a canonical
+        ``metric:<hash>`` definition (rxdo.9.1) from the process-wide
+        default registry (``polylogue/insights/measurement/
+        registered_metrics.py``) -- identity/schema resolution only, not
+        execution (no aggregation engine exists yet; see polylogue-9l5.7).
         """
         normalized = _object_ref(ref)
         session_id = normalized.removeprefix("session:") if normalized.startswith("session:") else None
+        plan_name = normalized.removeprefix("cost-outlook:") if normalized.startswith("cost-outlook:") else None
+        metric_id = normalized.removeprefix("metric:") if normalized.startswith("metric:") else None
 
         async def run() -> str:
+            if plan_name is not None:
+                return await _cost_outlook_payload(hooks, plan_name=plan_name, method=projection)
+            if metric_id is not None:
+                return _metric_definition_payload(hooks, metric_id)
             if projection == "events" and session_id is not None:
                 events = await hooks.get_polylogue().get_session_events(session_id)
                 if events is None:
diff --git a/polylogue/mcp/server_prompts.py b/polylogue/mcp/server_prompts.py
index cc149b984e..b7f1a6e0c7 100644
--- a/polylogue/mcp/server_prompts.py
+++ b/polylogue/mcp/server_prompts.py
@@ -542,10 +542,11 @@ async def cost_of(since: str = "30d", limit: int = 10) -> str:
         return f"""Report cost/usage for the last {since}, with honest accounting.
 
 Call sequence:
-1. cost_rollups(since="{since}") — aggregate spend by model.
-2. session_costs(since="{since}", limit={limit}) — top sessions by cost.
-3. provider_usage(detail="summary") — usage accounting diagnostics without billing estimates.
-4. For the current repo's sessions: search(query={repository_query!r}) then session_costs(session_id=<id>) per hit — cost tools have no repo filter.
+1. get(ref="cost-outlook:<plan_name>") — current billing-cycle projection for a configured subscription plan (burn rate, quota pressure, overage).
+2. status(scope="archive", include=["provider_usage"]) — usage accounting diagnostics without billing estimates.
+3. For the current repo's sessions: query(expression={repository_query!r}) for hits, then get(ref="session:<id>") per hit for per-session cost detail.
+
+Note: per-model/per-session cost rollup listing (formerly the retired cost_rollups/session_costs tools) has no MCP surface yet -- use the CLI `polylogue analyze --insight cost-rollups` / `--insight costs` commands for that (tracked as remaining scope on polylogue-hg97).
 
 Rules:
 - cost_usd is API-list-equivalent; on subscription plans report the subscription-credit view separately (cache reads are ~free on Claude Max/Pro).
diff --git a/polylogue/storage/sqlite/archive_tiers/user_write.py b/polylogue/storage/sqlite/archive_tiers/user_write.py
index d8e6c7a9d7..d6e71a3510 100644
--- a/polylogue/storage/sqlite/archive_tiers/user_write.py
+++ b/polylogue/storage/sqlite/archive_tiers/user_write.py
@@ -17,7 +17,7 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import PurePosixPath, PureWindowsPath
-from typing import TYPE_CHECKING, Final
+from typing import TYPE_CHECKING, Any, Final, cast
 
 from polylogue.core.assertions import (
     AssertionContextPolicy,
@@ -584,6 +584,22 @@ class FindingAssertion:
     evaluation_ref: str | None = None
     frame_ref: str | None = None
     public_claim: PublicClaimDeclaration | None = None
+    controls: Sequence[Mapping[str, JSONValue]] = ()
+    """Paired negative controls (rxdo.9.7, mechanism G). Each entry mirrors
+    :class:`~polylogue.insights.judgment.controls.NegativeControl`'s fields
+    (``control_kind``, ``query_ref``, ``result_ref``, ``matching_variables``,
+    ``expected_null``, ``confounds_checked``) plus ``observed_null_held``
+    (bool) -- the detector/analyst runs the control comparison and declares
+    both the control and its observed outcome together at write time; no
+    execution engine reruns it at read time. Rejected (unmatched/confounded)
+    controls fail the whole finding write closed rather than being silently
+    dropped or stored unvalidated -- see :func:`_finding_value`."""
+    control_frame_variables: Sequence[str] = ()
+    """Frame variables the finding's claim varies over -- required for
+    :func:`~polylogue.insights.judgment.controls.validate_control` to check
+    that a declared matched-shape control isolates the right mechanism, or
+    that an ``unrelated_cohort`` control declares every frame variable as a
+    checked confound."""
 
 
 def upsert_suppression(
@@ -1590,9 +1606,56 @@ def _finding_value(finding: FindingAssertion) -> dict[str, object]:
         value["frame_ref"] = _validate_finding_ref(finding.frame_ref, field="frame_ref")
     if finding.public_claim is not None:
         value["public_claim"] = _public_claim_value(finding.public_claim)
+    if finding.controls:
+        value["controls"] = _finding_controls_value(finding)
     return value
 
 
+def _finding_controls_value(finding: FindingAssertion) -> list[dict[str, object]]:
+    """Validate declared negative controls, failing the write closed on any rejection.
+
+    Reuses :func:`~polylogue.insights.judgment.controls.validate_control`
+    (rxdo.9.7, mechanism G) -- a control is stored only if it isolates the
+    challenged mechanism (matched-shape) or declares every frame variable a
+    checked confound (``unrelated_cohort``); a deliberately divergent
+    baseline never silently passes as a control.
+    """
+    from polylogue.insights.judgment.controls import NegativeControl, validate_control
+
+    frame_variables = tuple(finding.control_frame_variables)
+    validated: list[dict[str, object]] = []
+    for raw in finding.controls:
+        try:
+            control = NegativeControl(
+                control_kind=cast(Any, raw["control_kind"]),
+                query_ref=_validate_finding_ref(cast(str, raw["query_ref"]), field="controls.query_ref"),
+                result_ref=_validate_finding_ref(cast(str, raw["result_ref"]), field="controls.result_ref"),
+                matching_variables=tuple(cast(Sequence[str], raw.get("matching_variables", ()))),
+                expected_null=cast(str, raw["expected_null"]),
+                confounds_checked=tuple(cast(Sequence[str], raw.get("confounds_checked", ()))),
+            )
+        except (KeyError, ValueError) as exc:
+            raise ValueError(f"malformed finding control: {exc}") from exc
+        outcome = validate_control(control, claim_frame_variables=frame_variables)
+        if not outcome.accepted:
+            raise ValueError(f"finding control rejected: {outcome.reason}")
+        observed_null_held = raw.get("observed_null_held")
+        if not isinstance(observed_null_held, bool):
+            raise ValueError("finding control requires a boolean observed_null_held")
+        validated.append(
+            {
+                "control_kind": control.control_kind,
+                "query_ref": control.query_ref,
+                "result_ref": control.result_ref,
+                "matching_variables": list(control.matching_variables),
+                "expected_null": control.expected_null,
+                "confounds_checked": list(control.confounds_checked),
+                "observed_null_held": observed_null_held,
+            }
+        )
+    return validated
+
+
 def upsert_findings_as_assertions(
     conn: sqlite3.Connection,
     findings: Sequence[FindingAssertion],
@@ -1626,6 +1689,12 @@ def upsert_findings_as_assertions(
             public_refs = public_claim.get("public_evidence_refs")
             if isinstance(public_refs, list):
                 evidence_refs.extend(str(ref) for ref in public_refs)
+        controls_value = value.get("controls")
+        if isinstance(controls_value, list):
+            for control_entry in controls_value:
+                if isinstance(control_entry, dict):
+                    evidence_refs.append(str(control_entry["query_ref"]))
+                    evidence_refs.append(str(control_entry["result_ref"]))
         evidence_refs = sorted(set(evidence_refs))
         detector_ref = _validate_finding_ref(finding.detector_ref, field="detector_ref")
         assertion_id = assertion_id_for_finding(
diff --git a/tests/unit/api/test_facade_contracts.py b/tests/unit/api/test_facade_contracts.py
index ff6cf345dc..aaf86d3ef9 100644
--- a/tests/unit/api/test_facade_contracts.py
+++ b/tests/unit/api/test_facade_contracts.py
@@ -275,6 +275,12 @@
         "compare_sessions",
         "find_similar_sessions_by_metadata",
         "correlate_sessions",
+        # Comparative judgment storage (rxdo.9.6/.9.7/.9.11/.9.12), wired
+        # into the real `polylogue compare` CLI command (tests/unit/cli/
+        # test_compare_command.py) and the finding-controls read path
+        # (test_resolve_ref_renders_finding_claim_with_controls above).
+        "record_comparative_judgment",
+        "list_comparative_judgments",
     }
 )
 
@@ -2938,6 +2944,106 @@ async def test_resolve_ref_returns_finding_provenance_payload(tmp_path: Path) ->
         await archive.close()
 
 
+async def test_resolve_ref_renders_finding_claim_with_controls(tmp_path: Path) -> None:
+    """polylogue-rxdo.9.7: a finding with declared negative controls renders claim-vs-control.
+
+    Before this, ``ClaimWithControls`` (rxdo.9.7, mutation-tested) had zero
+    callers outside its own unit tests. This is the real production route:
+    a finding written through the real storage writer, resolved through the
+    real ``Polylogue.resolve_ref`` facade.
+    """
+    from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+    from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+    from polylogue.storage.sqlite.archive_tiers.user_write import FindingAssertion, upsert_findings_as_assertions
+
+    archive = _archive(tmp_path)
+    try:
+        user_db = archive.config.archive_root / "user.db"
+        initialize_archive_database(user_db, ArchiveTier.USER)
+        with sqlite3.connect(user_db) as conn:
+            envelopes = upsert_findings_as_assertions(
+                conn,
+                [
+                    FindingAssertion(
+                        claim_key="controlled-claim",
+                        target_ref="query:controlled-claim-v1",
+                        body_text="A claim backed by a passing negative control.",
+                        finding_kind="measure",
+                        statistic={"op": "rate", "value": 0.18, "unit": "fraction"},
+                        n=200,
+                        query_ref="query:controlled-claim-v1",
+                        result_set_ref="result-set:controlled-claim-run",
+                        detector_ref="agent:controls-detector",
+                        controls=(
+                            {
+                                "control_kind": "shifted_window",
+                                "query_ref": "query:controlled-claim-control",
+                                "result_ref": "result-set:controlled-claim-control",
+                                "matching_variables": ["repo"],
+                                "expected_null": "no rate change outside the treatment window",
+                                "observed_null_held": True,
+                            },
+                        ),
+                        control_frame_variables=("repo",),
+                    )
+                ],
+                now_ms=1,
+            )
+            conn.commit()
+        assertion_id = envelopes[0].assertion_id
+
+        passing = await archive.resolve_ref(f"finding:{assertion_id}")
+        assert passing.payload is not None
+        assert passing.payload["rank_tier"] == "controlled_pass"
+        assert passing.payload["downgraded"] is False
+        assert passing.payload["controls"][0]["observed_null_held"] is True
+        # The control's own query/result refs are ad-hoc (not backed by real
+        # archived query/result-set rows), so staleness caveats independently
+        # about evidence resolvability are expected here; the assertion of
+        # interest is that no *downgrade* caveat is added for a passing control.
+        assert not any("downgraded" in caveat for caveat in passing.caveats)
+
+        with sqlite3.connect(user_db) as conn:
+            failing_envelopes = upsert_findings_as_assertions(
+                conn,
+                [
+                    FindingAssertion(
+                        claim_key="downgraded-claim",
+                        target_ref="query:downgraded-claim-v1",
+                        body_text="A claim whose negative control failed.",
+                        finding_kind="measure",
+                        statistic={"op": "rate", "value": 0.18, "unit": "fraction"},
+                        n=200,
+                        query_ref="query:downgraded-claim-v1",
+                        result_set_ref="result-set:downgraded-claim-run",
+                        detector_ref="agent:controls-detector",
+                        controls=(
+                            {
+                                "control_kind": "shifted_window",
+                                "query_ref": "query:downgraded-claim-control",
+                                "result_ref": "result-set:downgraded-claim-control",
+                                "matching_variables": ["repo"],
+                                "expected_null": "no rate change outside the treatment window",
+                                "observed_null_held": False,
+                            },
+                        ),
+                        control_frame_variables=("repo",),
+                    )
+                ],
+                now_ms=1,
+            )
+            conn.commit()
+        failing_assertion_id = failing_envelopes[0].assertion_id
+
+        downgraded = await archive.resolve_ref(f"finding:{failing_assertion_id}")
+        assert downgraded.payload is not None
+        assert downgraded.payload["rank_tier"] == "controlled_fail"
+        assert downgraded.payload["downgraded"] is True
+        assert any("downgraded" in caveat for caveat in downgraded.caveats)
+    finally:
+        await archive.close()
+
+
 def _delegation_parent_session(*, provider_session_id: str, with_dispatch: bool) -> ParsedSession:
     """Ingest-shaped parent fixture: writes real session/message/block rows
     through the live archive writer (``ArchiveStore.write_parsed`` ->
diff --git a/tests/unit/cli/test_compare_command.py b/tests/unit/cli/test_compare_command.py
new file mode 100644
index 0000000000..80c65720f2
--- /dev/null
+++ b/tests/unit/cli/test_compare_command.py
@@ -0,0 +1,118 @@
+"""``polylogue compare`` real production route (rxdo.9.6/.9.7/.9.11/.9.12).
+
+Before this, ``blind_items``/``BlindingReceipt``, ``ClaimWithControls``,
+``compute_calibration``, and the storage chokepoint
+``upsert_comparative_judgment_assertion`` had zero production callers -- only
+their own unit tests and, for ``blind_items``, an internal caller
+(``ElicitationSession``) that itself had no production caller either. This
+test exercises the real CLI command against a real archive: no mocked
+facade, no test double for the storage layer.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from click.testing import CliRunner
+
+from polylogue.cli.click_app import cli
+
+
+def test_compare_without_verdict_prints_blinded_pair_and_records_nothing(
+    cli_workspace: dict[str, Path],
+) -> None:
+    result = CliRunner().invoke(
+        cli,
+        [
+            "compare",
+            "--left",
+            "session:codex:a",
+            "--left-field",
+            "model=gpt-5",
+            "--right",
+            "session:codex:b",
+            "--right-field",
+            "model=claude-sonnet-5",
+            "--dimension",
+            "quality",
+            "--rubric",
+            "quality-v1",
+            "--json",
+        ],
+    )
+    assert result.exit_code == 0, result.output
+    body = json.loads(result.output)
+    assert len(body["items"]) == 2
+    # model is a masked provenance field -- must not leak into the blinded view.
+    for item in body["items"]:
+        assert "model" not in item["visible_fields"]
+        assert "ref" in item["visible_fields"]
+    assert body["receipt"]["revealed_at_ms"] is None
+
+
+def test_compare_with_verdict_records_and_is_readable_via_calibration(
+    cli_workspace: dict[str, Path],
+) -> None:
+    runner = CliRunner()
+    record = runner.invoke(
+        cli,
+        [
+            "compare",
+            "--left",
+            "session:codex:a",
+            "--right",
+            "session:codex:b",
+            "--dimension",
+            "quality",
+            "--rubric",
+            "quality-v1",
+            "--verdict",
+            "prefer_left",
+            "--actor-ref",
+            "agent:worker",
+            "--exec-context-id",
+            "ctx:1",
+            "--json",
+        ],
+    )
+    assert record.exit_code == 0, record.output
+    recorded = json.loads(record.output)
+    assert recorded["verdict"] == "prefer_left"
+    assert recorded["revealed"]["left"]["ref"] == "session:codex:a"
+
+    # A second recording of the same operator (gold) judgment on the same
+    # comparison lets --calibration compute a real agreement rate against it.
+    gold = runner.invoke(
+        cli,
+        [
+            "compare",
+            "--left",
+            "session:codex:a",
+            "--right",
+            "session:codex:b",
+            "--dimension",
+            "quality",
+            "--rubric",
+            "quality-v1",
+            "--verdict",
+            "prefer_left",
+            "--actor-ref",
+            "user:local",
+            "--exec-context-id",
+            "ctx:gold",
+            "--json",
+        ],
+    )
+    assert gold.exit_code == 0, gold.output
+
+    calibration = runner.invoke(
+        cli,
+        ["compare", "--calibration", "--gold-actor", "user:local", "--json"],
+    )
+    assert calibration.exit_code == 0, calibration.output
+    reports = json.loads(calibration.output)
+    worker_reports = [r for r in reports if r["actor_ref"] == "agent:worker"]
+    assert len(worker_reports) == 1
+    assert worker_reports[0]["agreement_rate"] == 1.0
+    assert worker_reports[0]["n_gold_overlap"] == 1
diff --git a/tests/unit/cost/test_contract_suite.py b/tests/unit/cost/test_contract_suite.py
index 464179e8f4..c67c89f219 100644
--- a/tests/unit/cost/test_contract_suite.py
+++ b/tests/unit/cost/test_contract_suite.py
@@ -483,24 +483,20 @@ def test_api_cost_outlook_returns_typed_cycle_outlook() -> None:
         assert key in payload, f"missing key {key!r} in outlook envelope"
 
 
-@pytest.mark.xfail(
-    reason=(
-        "The six-tool MCP cutover (polylogue-t46.8, #3095) retired the standalone "
-        "cost_outlook tool without a replacement -- cost/usage rollup MCP wiring "
-        "needs a design decision (point-fix vs. generic re-hosting of the whole "
-        "INSIGHT_REGISTRY cost/usage family), tracked in polylogue-hg97."
-    ),
-    raises=KeyError,
-    strict=True,
-)
 @pytest.mark.asyncio
-async def test_mcp_cost_outlook_tool_uses_shared_envelope() -> None:
-    """The MCP ``cost_outlook`` tool serializes the same typed envelope.
-
-    The MCP tool is a leaf adapter — it must not redefine the cost outlook
-    shape. This test builds the MCP server, mocks the facade, calls the
-    ``cost_outlook`` tool, and asserts the JSON envelope contains the typed
-    fields produced by the cost engine.
+async def test_mcp_get_cost_outlook_ref_uses_shared_envelope() -> None:
+    """The consolidated ``get`` tool resolves ``cost-outlook:<plan>`` refs.
+
+    The six-tool MCP cutover (polylogue-t46.8, #3095) retired the standalone
+    ``cost_outlook`` tool without a replacement; polylogue-hg97 designed and
+    landed the replacement as a ``get(ref="cost-outlook:<plan>", ...)``
+    resolution (rather than an 11th top-level tool, which would contradict
+    the consolidated-tool architecture) -- see ``_cost_outlook_payload`` in
+    ``polylogue/mcp/server_cutover.py``. The MCP tool is a leaf adapter — it
+    must not redefine the cost outlook shape. This test builds the MCP
+    server, mocks the facade, calls ``get`` with a cost-outlook ref, and
+    asserts the JSON envelope contains the typed fields produced by the cost
+    engine.
     """
     import asyncio as _asyncio
     import json as _json
@@ -520,16 +516,46 @@ async def test_mcp_cost_outlook_tool_uses_shared_envelope() -> None:
         facade_mock.cost_outlook = AsyncMock(return_value=outlook)
         mock_get_polylogue.return_value = facade_mock
         raw = await invoke_surface_async(
-            server._tool_manager._tools["cost_outlook"].fn,
-            plan="claude-pro",
-            method="linear",
+            server._tool_manager._tools["get"].fn,
+            ref="cost-outlook:claude-pro",
+            projection="linear",
         )
+        facade_mock.cost_outlook.assert_awaited_once()
 
     payload = _json.loads(raw)
-    # The MCP cost_outlook tool emits the typed CycleOutlook payload at the
-    # top level (matching the existing test_cost_outlook_tool.py contract).
+    # The get() cost-outlook resolution emits the typed CycleOutlook payload
+    # at the top level (matching the retired standalone-tool contract).
     for key in ("plan_name", "window", "projection_method", "quota_pressure"):
-        assert key in payload, f"MCP cost_outlook payload missing {key!r}"
+        assert key in payload, f"MCP get(cost-outlook:...) payload missing {key!r}"
+
+
+@pytest.mark.asyncio
+async def test_mcp_get_cost_outlook_unknown_plan_reports_typed_error() -> None:
+    """An unknown plan name resolves to a typed invalid_argument error, not a KeyError/traceback."""
+    import asyncio as _asyncio
+    import json as _json
+
+    from tests.infra.mcp import MCPServerUnderTest, invoke_surface_async, make_polylogue_mock
+
+    _asyncio.set_event_loop_policy(None)
+    from polylogue.cost.plans import PlanLookupError
+    from polylogue.mcp.server import build_server
+    from tests.infra.mcp import ALL_CAPABILITIES
+
+    server = build_server(capabilities=ALL_CAPABILITIES)
+    assert isinstance(server, MCPServerUnderTest)
+
+    with patch("polylogue.mcp.server._get_polylogue") as mock_get_polylogue:
+        facade_mock = make_polylogue_mock()
+        facade_mock.cost_outlook = AsyncMock(side_effect=PlanLookupError("unknown plan: nope"))
+        mock_get_polylogue.return_value = facade_mock
+        raw = await invoke_surface_async(
+            server._tool_manager._tools["get"].fn,
+            ref="cost-outlook:nope",
+        )
+
+    payload = _json.loads(raw)
+    assert payload.get("code") == "invalid_argument", payload
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/unit/insights/measurement/test_registered_metrics.py b/tests/unit/insights/measurement/test_registered_metrics.py
new file mode 100644
index 0000000000..17ff801cc6
--- /dev/null
+++ b/tests/unit/insights/measurement/test_registered_metrics.py
@@ -0,0 +1,23 @@
+"""The process-wide default metric registry has a real registered definition (rxdo.9.1)."""
+
+from __future__ import annotations
+
+from polylogue.insights.measurement.registered_metrics import DEFAULT_METRIC_REGISTRY, SESSION_COST_USD_METRIC
+
+
+def test_default_registry_resolves_the_session_cost_metric_by_name() -> None:
+    resolved = DEFAULT_METRIC_REGISTRY.resolve("session_cost_usd")
+    assert resolved is not None
+    assert resolved.ref == SESSION_COST_USD_METRIC.ref
+
+
+def test_default_registry_resolves_the_session_cost_metric_by_hash() -> None:
+    resolved = DEFAULT_METRIC_REGISTRY.get(SESSION_COST_USD_METRIC.ref)
+    assert resolved is not None
+    assert resolved.construct == SESSION_COST_USD_METRIC.construct
+
+
+def test_session_cost_metric_declares_mixed_provenance_honestly() -> None:
+    """Session cost blends provider-reported and catalog-estimated lanes -- must be declared, not silent."""
+    assert SESSION_COST_USD_METRIC.provenance_mixing == "mixed-declared"
+    assert set(SESSION_COST_USD_METRIC.measurement_authority) == {"provider-reported", "catalog-estimated"}
diff --git a/tests/unit/mcp/test_metric_ref_resolution.py b/tests/unit/mcp/test_metric_ref_resolution.py
new file mode 100644
index 0000000000..2afc381203
--- /dev/null
+++ b/tests/unit/mcp/test_metric_ref_resolution.py
@@ -0,0 +1,48 @@
+"""MCP ``get`` tool resolves ``metric:<hash-or-name>`` refs (polylogue-rxdo.9.1).
+
+``MetricDefinition``/``MetricRegistry`` (PR #2888, merged) had zero
+production callers -- the corrective AC's second consumer path depends on
+``polylogue-9l5.7``'s statistics registry, which remains unstarted (that
+whole registry/composition epic is out of scope for this wiring pass). This
+test proves the bounded, honest slice that IS wired: the process-wide
+``DEFAULT_METRIC_REGISTRY`` (``polylogue/insights/measurement/
+registered_metrics.py``) resolves through the real MCP ``get`` tool, not
+just its own unit tests.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from polylogue.insights.measurement.registered_metrics import SESSION_COST_USD_METRIC
+from tests.infra.mcp import MCPServerUnderTest, invoke_surface
+from tests.unit.mcp.test_contract_evidence import _seeded_runtime_services
+
+
+def test_mcp_get_resolves_metric_ref_by_registered_name(mcp_server: MCPServerUnderTest, tmp_path: Path) -> None:
+    archive_root = tmp_path / "archive"
+    with _seeded_runtime_services(archive_root):
+        result = invoke_surface(mcp_server._tool_manager._tools["get"].fn, ref="metric:session_cost_usd")
+
+    body = json.loads(result)
+    assert body["ref"] == SESSION_COST_USD_METRIC.ref
+    assert body["definition"]["construct"] == SESSION_COST_USD_METRIC.construct
+
+
+def test_mcp_get_resolves_metric_ref_by_content_hash(mcp_server: MCPServerUnderTest, tmp_path: Path) -> None:
+    archive_root = tmp_path / "archive"
+    with _seeded_runtime_services(archive_root):
+        result = invoke_surface(mcp_server._tool_manager._tools["get"].fn, ref=SESSION_COST_USD_METRIC.ref)
+
+    body = json.loads(result)
+    assert body["definition"]["output_schema"] == "usd:float"
+
+
+def test_mcp_get_unknown_metric_ref_returns_typed_not_found(mcp_server: MCPServerUnderTest, tmp_path: Path) -> None:
+    archive_root = tmp_path / "archive"
+    with _seeded_runtime_services(archive_root):
+        result = invoke_surface(mcp_server._tool_manager._tools["get"].fn, ref="metric:no-such-metric")
+
+    body = json.loads(result)
+    assert body.get("code") == "not_found", body
diff --git a/tests/unit/mcp/test_reference_query_pipeline.py b/tests/unit/mcp/test_reference_query_pipeline.py
new file mode 100644
index 0000000000..9acd4f47d7
--- /dev/null
+++ b/tests/unit/mcp/test_reference_query_pipeline.py
@@ -0,0 +1,128 @@
+"""MCP ``query`` tool resolves ``from query:<hash>|result-set:<id>`` pipelines (polylogue-rxdo.6).
+
+Before this, ``ReferenceQueryPipeline``/``RefOperand``/``parse_reference_query_pipeline``
+and the real ``DurableRefResolver``/``ArchiveCanonicalPlanEvaluator`` planner
+seam (both landed as production implementations in PR #2899) had zero callers
+anywhere in ``polylogue/cli/*.py``, ``polylogue/mcp/*.py``, or
+``polylogue/daemon/*.py`` -- the compatibility selector in
+``archive/query/expression.py:compile_expression`` unconditionally hard-erred
+on any ``from <ref>`` pipeline. These tests exercise the real production
+route wired into the MCP ``query`` tool (``_resolve_reference_query_pipeline``
+in ``polylogue/mcp/server_cutover.py``): a real archive, a real durable
+``query:<hash>`` object, and the actual planner/resolver classes -- no test
+double stands in for either.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+
+from polylogue.archive.message.roles import Role
+from polylogue.core.enums import BlockType, Provider
+from polylogue.core.query_identity import JsonValue
+from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession
+from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+from polylogue.storage.sqlite.query_objects import QueryObject, put_query
+from tests.infra.mcp import MCPServerUnderTest, invoke_surface
+from tests.unit.mcp.test_contract_evidence import _seeded_runtime_services
+
+
+def _seed_archive(archive_root: Path) -> None:
+    archive_root.mkdir(parents=True, exist_ok=True)
+    with ArchiveStore(archive_root) as archive:
+        for provider, native_id, title in (
+            (Provider.CODEX, "codex-1", "codex session"),
+            (Provider.CLAUDE_CODE, "claude-1", "claude session"),
+        ):
+            archive.write_parsed(
+                ParsedSession(
+                    source_name=provider,
+                    provider_session_id=native_id,
+                    title=title,
+                    created_at="2026-01-01T00:00:00+00:00",
+                    updated_at="2026-01-01T00:01:00+00:00",
+                    messages=[
+                        ParsedMessage(
+                            provider_message_id=f"{native_id}-m1",
+                            role=Role.USER,
+                            text="hello",
+                            timestamp="2026-01-01T00:00:00+00:00",
+                            blocks=[ParsedContentBlock(type=BlockType.TEXT, text="hello")],
+                        )
+                    ],
+                )
+            )
+    initialize_archive_database(archive_root / "user.db", ArchiveTier.USER)
+    initialize_archive_database(archive_root / "ops.db", ArchiveTier.OPS)
+
+
+def _origin_query(conn: sqlite3.Connection, *, origin: str) -> QueryObject:
+    ast: dict[str, JsonValue] = {
+        "kind": "field",
+        "field": "origin",
+        "op": "=",
+        "values": [origin],
+    }
+    return put_query(conn, ast, grain="session", lane="dialogue", rank_policy="mixed", created_at_ms=1)
+
+
+def test_mcp_query_resolves_from_query_reference(mcp_server: MCPServerUnderTest, tmp_path: Path) -> None:
+    archive_root = tmp_path / "archive"
+    _seed_archive(archive_root)
+    with sqlite3.connect(archive_root / "user.db") as conn:
+        query = _origin_query(conn, origin="codex-session")
+        conn.commit()
+
+    with _seeded_runtime_services(archive_root):
+        result = invoke_surface(
+            mcp_server._tool_manager._tools["query"].fn,
+            expression=f"from query:{query.query_hash}",
+        )
+
+    body = json.loads(result)
+    assert body["source"] == f"query:{query.query_hash}"
+    assert body["grain"] == "session"
+    assert body["member_count"] == 1
+    assert len(body["members"]) == 1
+    assert body["members"][0].startswith("session:codex-session:")
+    assert body["truncated"] is False
+
+
+def test_mcp_query_from_unknown_query_hash_returns_typed_not_found(
+    mcp_server: MCPServerUnderTest, tmp_path: Path
+) -> None:
+    archive_root = tmp_path / "archive"
+    _seed_archive(archive_root)
+
+    with _seeded_runtime_services(archive_root):
+        result = invoke_surface(
+            mcp_server._tool_manager._tools["query"].fn,
+            expression="from query:0000000000000000000000000000000000000000000000000000000000000000",
+        )
+
+    body = json.loads(result)
+    assert body.get("code") == "not_found", body
+
+
+def test_mcp_query_from_reference_with_stages_returns_typed_not_implemented(
+    mcp_server: MCPServerUnderTest, tmp_path: Path
+) -> None:
+    """Stage composition after the root operand is honestly unimplemented, not silently dropped or crashed."""
+    archive_root = tmp_path / "archive"
+    _seed_archive(archive_root)
+    with sqlite3.connect(archive_root / "user.db") as conn:
+        query = _origin_query(conn, origin="codex-session")
+        conn.commit()
+
+    with _seeded_runtime_services(archive_root):
+        result = invoke_surface(
+            mcp_server._tool_manager._tools["query"].fn,
+            expression=f"from query:{query.query_hash} | count",
+        )
+
+    body = json.loads(result)
+    assert body.get("code") == "not_implemented", body
diff --git a/tests/unit/storage/test_archive_tiers_assertions.py b/tests/unit/storage/test_archive_tiers_assertions.py
index 79a2bbac7d..0222ab1a6a 100644
--- a/tests/unit/storage/test_archive_tiers_assertions.py
+++ b/tests/unit/storage/test_archive_tiers_assertions.py
@@ -12,6 +12,7 @@
 from polylogue.archive.message.roles import Role
 from polylogue.archive.session.domain_models import Session
 from polylogue.core.enums import Origin
+from polylogue.core.json import JSONValue
 from polylogue.core.types import SessionId
 from polylogue.insights.transforms import compile_session_digest
 from polylogue.storage.sqlite.archive_tiers import user_write
@@ -1892,6 +1893,81 @@ def test_upsert_findings_rejects_incomplete_delta_and_unresolved_ref_shapes(tmp_
         conn.close()
 
 
+def _finding_with_control(**control_overrides: JSONValue) -> FindingAssertion:
+    control: dict[str, JSONValue] = {
+        "control_kind": "shifted_window",
+        "query_ref": "query:control-window",
+        "result_ref": "result-set:control-window",
+        "matching_variables": ["repo"],
+        "expected_null": "no rate change outside the treatment window",
+        "observed_null_held": True,
+    }
+    control.update(control_overrides)
+    return FindingAssertion(
+        claim_key="tool-failure-rate-with-control",
+        target_ref="query:tool-failure-rate-v1",
+        body_text="The failure rate increased from the pre-registered baseline.",
+        finding_kind="measure",
+        statistic={"op": "rate", "value": 0.18, "unit": "fraction"},
+        n=200,
+        query_ref="query:tool-failure-rate-v1",
+        result_set_ref="result-set:tool-failure-rate-run-2",
+        detector_ref="insight:tool-failure-detector@v1",
+        controls=(control,),
+        control_frame_variables=("repo",),
+    )
+
+
+def test_upsert_finding_with_matched_control_stores_validated_observed_outcome(tmp_path: Path) -> None:
+    """A matched-shape control that isolates the frame variable is accepted and stored (rxdo.9.7)."""
+    conn = _connect(tmp_path / "user.db")
+    try:
+        finding = _finding_with_control()
+        written = upsert_findings_as_assertions(conn, [finding], now_ms=1_700_000_000_000)
+        assert len(written) == 1
+        value = written[0].value
+        assert isinstance(value, dict)
+        stored_controls = value["controls"]
+        assert stored_controls == [
+            {
+                "control_kind": "shifted_window",
+                "query_ref": "query:control-window",
+                "result_ref": "result-set:control-window",
+                "matching_variables": ["repo"],
+                "expected_null": "no rate change outside the treatment window",
+                "confounds_checked": [],
+                "observed_null_held": True,
+            }
+        ]
+        # Control query/result refs are resolvable evidence, not opaque metadata.
+        assert "query:control-window" in written[0].evidence_refs
+        assert "result-set:control-window" in written[0].evidence_refs
+    finally:
+        conn.close()
+
+
+def test_upsert_finding_rejects_confounded_unrelated_cohort_control(tmp_path: Path) -> None:
+    """A deliberately divergent baseline that leaves the claim's frame variable unchecked fails closed."""
+    conn = _connect(tmp_path / "user.db")
+    try:
+        finding = _finding_with_control(control_kind="unrelated_cohort", confounds_checked=[])
+        with pytest.raises(ValueError, match="control rejected"):
+            upsert_findings_as_assertions(conn, [finding])
+    finally:
+        conn.close()
+
+
+def test_upsert_finding_rejects_control_without_matching_variables(tmp_path: Path) -> None:
+    """A matched-shape control declaring no matching variables isolates nothing -- rejected, not silently stored."""
+    conn = _connect(tmp_path / "user.db")
+    try:
+        finding = _finding_with_control(matching_variables=[])
+        with pytest.raises(ValueError, match="control rejected"):
+            upsert_findings_as_assertions(conn, [finding])
+    finally:
+        conn.close()
+
+
 def test_upsert_assertion_owned_transaction_rolls_back_on_write_failure(tmp_path: Path) -> None:
     """A failed standalone upsert cannot leave an owned transaction or partial row."""