feat(dashboard-testing): pin published baselines in schedules

2026-09-25 11:44:44 +03:00
parent f9d2f8d9bc
commit 354c6684d8
21 changed files with 803 additions and 45 deletions
--- a/backend/alembic/versions/0027_schedule_baseline.py
+++ b/backend/alembic/versions/0027_schedule_baseline.py
@@ -0,0 +1,29 @@
+# #region Migrations.ScenarioScheduleBaselineSelector [C:2] [TYPE Module] [SEMANTICS migration,scenario,schedule,baseline]
+# @BRIEF Store the explicit published baseline selector with the durable scenario schedule.
+# @RATIONALE The scheduler callback must reload this selector on every fire so a restored job
+#            admits baseline-backed revisions without resolving a reference URL or short key.
+# @REJECTED Resolving the short key during dispatch would allow an edited or expired permalink
+#           to change the scheduled run's reference identity.
+"""Persist baseline selectors on scenario schedules."""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0027_schedule_baseline"
+down_revision = "0026_maintenance_templates"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    columns = {column["name"] for column in sa.inspect(op.get_bind()).get_columns("scenario_schedules")}
+    if "baseline_set" not in columns:
+        op.add_column("scenario_schedules", sa.Column("baseline_set", sa.String(length=255), nullable=True))
+    if "baseline_set_version" not in columns:
+        op.add_column("scenario_schedules", sa.Column("baseline_set_version", sa.String(length=128), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("scenario_schedules", "baseline_set_version")
+    op.drop_column("scenario_schedules", "baseline_set")
+# #endregion Migrations.ScenarioScheduleBaselineSelector
--- a/backend/src/api/routes/dashboard_testing/scenario_automation.py
+++ b/backend/src/api/routes/dashboard_testing/scenario_automation.py
@@ -39,7 +39,7 @@ from __future__ import annotations
 from typing import Any

 from fastapi import APIRouter, Depends, Header, HTTPException, status
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator

 from src.dependencies import get_config_manager, get_current_user, get_db, get_scheduler_service, has_permission
 from src.models.scenario_automation import AutomationPolicy, ScenarioNotificationEvent, ScenarioRetentionDeletion, ScenarioSchedule, ScenarioTriggerRule
@@ -130,6 +130,8 @@ class ScheduleRequest(BaseModel):
    scenario_id: str
    environment_id: str
    cron_expr: str
+    baseline_set: str | None = None
+    baseline_set_version: str | None = None
    revision_policy: str = "current"
    revision_id: str | None = None
    timezone: str = "UTC"
@@ -138,6 +140,12 @@ class ScheduleRequest(BaseModel):
    max_instances: int = Field(default=1, ge=1)
    misfire_grace_time: int = Field(default=300, ge=0)
    enabled: bool = True
+
+    @model_validator(mode="after")
+    def baseline_selector_pair(self):
+        if bool((self.baseline_set or "").strip()) != bool((self.baseline_set_version or "").strip()):
+            raise ValueError("baseline_set and baseline_set_version must be provided together")
+        return self
 # #endregion Api.ScenarioAutomation.ScheduleRequest


--- a/backend/src/core/scheduler.py
+++ b/backend/src/core/scheduler.py
@@ -319,6 +319,8 @@ def _scheduled_idempotency_key(schedule_id: str, fired_at: datetime) -> str:
 #            cannot downgrade a target or dispatch pending_approval work.
 # @INVARIANT Idempotency is deterministic per (schedule_id, fired_at-second): a same-second
 #            double-fire replays the existing run through the unique key instead of duplicating it.
+# @INVARIANT A baseline-backed schedule reloads its explicit set/version from the durable row;
+#            reference URL state is frozen in the published catalog pin, never fetched at fire time.
 # @RATIONALE Module-level callback follows the same pattern as backup/validation callbacks so
 #            the persistent job store never serializes dependencies. Dependencies (runner, models)
 #            are resolved at execution time to avoid import cycles.
@@ -338,6 +340,10 @@ def execute_scheduled_scenario(schedule_id: str, scenario_id: str, revision_poli

        schedule_row = db.query(_ScenarioSchedule).filter(_ScenarioSchedule.id == schedule_id).first()
        owner_actor = (schedule_row.created_by if schedule_row is not None and schedule_row.created_by else "apscheduler")
+        # The row is authoritative across persistent APScheduler job restores. The published
+        # catalog carries frozen reference URL/filter evidence; do not resolve a short key here.
+        baseline_set = schedule_row.baseline_set if schedule_row is not None else None
+        baseline_set_version = schedule_row.baseline_set_version if schedule_row is not None else None

        resolved_revision_id = revision_id
        if revision_policy == "current":
@@ -363,6 +369,8 @@ def execute_scheduled_scenario(schedule_id: str, scenario_id: str, revision_poli
            idempotency_key=_scheduled_idempotency_key(schedule_id, datetime.now(UTC)),
            auto_advance=False,
            trigger_source="scheduled",
+            baseline_set=baseline_set,
+            baseline_set_version=baseline_set_version,
        )
        db.commit()
        logger.reason(
--- a/backend/src/mcp_server/tools_automation.py
+++ b/backend/src/mcp_server/tools_automation.py
@@ -15,7 +15,7 @@
 from typing import Any
 import hashlib
 import json
-from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt, StrictStr
+from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt, StrictStr, model_validator
 from src.core.auth.permission_utils import user_has_permission
 from src.core.auth.repository import AuthRepository
 from src.core.database import SessionLocal
@@ -44,12 +44,20 @@ class AutomationBase(BaseModel):

 class ScheduleInput(AutomationBase):
    cron_expr: StrictStr = Field(min_length=1, max_length=128)
+    baseline_set: StrictStr | None = Field(default=None, max_length=255)
+    baseline_set_version: StrictStr | None = Field(default=None, max_length=128)
    timezone: StrictStr = Field(default="UTC", max_length=64)
    missed_execution_policy: StrictStr = "skip"
    max_instances: StrictInt = Field(default=1, ge=1)
    misfire_grace_time: StrictInt = Field(default=300, ge=0)
    policy_id: StrictStr | None = None

+    @model_validator(mode="after")
+    def baseline_selector_pair(self):
+        if bool((self.baseline_set or "").strip()) != bool((self.baseline_set_version or "").strip()):
+            raise ValueError("baseline_set and baseline_set_version must be provided together")
+        return self
+

 class TriggerInput(AutomationBase):
    trigger: StrictStr = Field(min_length=1, max_length=64)
--- a/backend/src/models/scenario_automation.py
+++ b/backend/src/models/scenario_automation.py
@@ -27,6 +27,8 @@ class ScenarioSchedule(Base):
    scenario_id = Column(String(36), nullable=False, index=True)
    revision_policy = Column(String(16), nullable=False, default="current")
    revision_id = Column(String(36), nullable=True)
+    baseline_set = Column(String(255), nullable=True)
+    baseline_set_version = Column(String(128), nullable=True)
    environment_id = Column(String(128), nullable=False)
    cron_expr = Column(String(128), nullable=False)
    timezone = Column(String(64), nullable=False, default="UTC")
--- a/backend/src/services/dashboard_testing/execution/publication_receipt_binding.py
+++ b/backend/src/services/dashboard_testing/execution/publication_receipt_binding.py
@@ -0,0 +1,132 @@
+# #region ScenarioExecution.PublicationReceiptBinding [C:4] [TYPE Module] [SEMANTICS publish,git,receipt,pin,source]
+# @BRIEF Bind immutable Git catalog bytes to the durable server publication receipt before pin resolution.
+# @RELATION DEPENDS_ON -> [ScenarioExecution.PublishedCatalogSource]
+# @RELATION DEPENDS_ON -> [Models.ScenarioExecution.PublicationOperation]
+# @RELATION DEPENDS_ON -> [BaselineEngine.CatalogRevisionLog]
+# @RELATION CALLED_BY -> [ScenarioExecution.Runner.Start]
+# @PRE The deployment source is configured and db contains the publication operation created by
+#      the server's reviewed Git publication worker.
+# @POST Returns a snapshot whose publication block contains the observed Git head and operation ID
+#       only when immutable Git bytes exactly match the published operation envelope and receipt,
+#       and the declared catalog digest equals the digest recomputed from entry_revisions.
+#       Missing, moved, ambiguous or inconsistent evidence returns None and blocks the run pin.
+# @INVARIANT Neither caller bytes nor the raw Git envelope may assert its own publication commit.
+#            The Git head is observed first, the catalog is fetched at that exact SHA, and the
+#            durable operation must independently identify that SHA and those exact bytes.
+# @RATIONALE A Git commit cannot contain its own SHA. The append-only operation row is the
+#            server-owned receipt; binding it to immutable bytes at the observed head supplies
+#            the resolver's publication fields without weakening its published guard.
+# @REJECTED Stamping the latest branch head or trusting publication fields inside the raw envelope
+#           was rejected: either can manufacture a pin for bytes never committed by this operation.
+from __future__ import annotations
+
+from copy import deepcopy
+import json
+import re
+from typing import Any
+from urllib.parse import quote
+
+import httpx
+from sqlalchemy.orm import Session
+
+from src.core.logger import logger
+from src.models.publication_operation import PublicationOperation
+from src.services.dashboard_testing.catalog_revision_log import _digest_revisions
+
+from .published_catalog_source import _source_config, load_published_catalog_bytes_from_config
+
+_SRC = "ScenarioExecution.PublicationReceiptBinding"
+_COMMIT = re.compile(r"[a-f0-9]{40}")
+
+
+# #region ScenarioExecution.PublicationReceiptBinding.Load [C:4] [TYPE Function] [SEMANTICS git,receipt,source,bind]
+# @BRIEF Observe the Git head, fetch catalog bytes at that commit, and bind their durable receipt.
+# @POST Returns a resolver-ready snapshot only for a unique published operation whose envelope,
+#       repository, branch, commit and verified receipt match the observed Git catalog.
+def load_receipted_published_catalog(
+    db: Session,
+    baseline_set: str,
+    baseline_set_version: str,
+    *,
+    transport: httpx.BaseTransport | None = None,
+) -> dict[str, Any] | None:
+    config = _source_config()
+    if config is None:
+        return None
+    try:
+        with httpx.Client(
+            base_url=config["base_url"],
+            headers={"Authorization": f"token {config['token']}"},
+            timeout=5.0,
+            transport=transport,
+        ) as client:
+            response = client.get(
+                f"/api/v1/repos/{config['repo']}/branches/{quote(config['ref'], safe='')}"
+            )
+        if response.status_code != 200:
+            return None
+        branch = response.json()
+        commit = branch.get("commit") if isinstance(branch, dict) else None
+        head = (commit.get("id") or commit.get("sha")) if isinstance(commit, dict) else None
+        if not isinstance(head, str) or _COMMIT.fullmatch(head) is None:
+            return None
+    except (httpx.HTTPError, ValueError, TypeError):
+        return None
+
+    raw = load_published_catalog_bytes_from_config(
+        baseline_set, baseline_set_version, transport=transport, ref_override=head,
+    )
+    if raw is None:
+        return None
+    try:
+        snapshot = json.loads(raw)
+    except (UnicodeDecodeError, ValueError):
+        return None
+    if not isinstance(snapshot, dict):
+        return None
+    revision = snapshot.get("catalog_revision")
+    if not isinstance(revision, dict) or not isinstance(revision.get("publication"), dict):
+        return None
+    digest = snapshot.get("catalog_digest")
+    entries = revision.get("entry_revisions")
+    if (not isinstance(digest, str) or not isinstance(entries, list)
+            or revision.get("catalog_digest") != digest or _digest_revisions(entries) != digest):
+        return None
+    operations = db.query(PublicationOperation).filter(
+        PublicationOperation.state == "published",
+        PublicationOperation.repository == config["repo"],
+        PublicationOperation.branch == config["ref"],
+        PublicationOperation.commit_sha == head,
+        PublicationOperation.catalog_digest == digest,
+    ).all()
+    matching = [
+        row for row in operations
+        if json.dumps(row.envelope, sort_keys=True, ensure_ascii=False).encode("utf-8") == raw
+        and (row.catalog_revision_id is None or row.catalog_revision_id == revision.get("catalog_revision_id"))
+        and isinstance(row.published_receipt, dict)
+        and row.published_receipt.get("head_after") == head
+        and row.published_receipt.get("commit_sha") == head
+    ]
+    if len(matching) != 1:
+        logger.explore(
+            "Git catalog has no unique matching publication receipt",
+            src=_SRC, error_code="BASELINE_NOT_PUBLISHED",
+            payload={"repository": config["repo"], "branch": config["ref"], "matches": len(matching)},
+        )
+        return None
+    publication = revision["publication"]
+    if publication.get("commit_hash") not in (None, head) or publication.get("published_receipt_id") not in (None, matching[0].id):
+        return None
+    bound = deepcopy(snapshot)
+    bound["catalog_revision"]["publication"].update({
+        "state": "published", "commit_hash": head, "published_receipt_id": matching[0].id,
+        "error_code": None,
+    })
+    logger.reflect("Git catalog bound to durable publication receipt", src=_SRC, payload={
+        "repository": config["repo"], "branch": config["ref"], "catalog_digest": digest,
+        "publication_id": matching[0].id,
+    })
+    return bound
+# #endregion ScenarioExecution.PublicationReceiptBinding.Load
+
+# #endregion ScenarioExecution.PublicationReceiptBinding
--- a/backend/src/services/dashboard_testing/execution/published_catalog_source.py
+++ b/backend/src/services/dashboard_testing/execution/published_catalog_source.py
@@ -89,20 +89,22 @@ def load_published_catalog(snapshot: dict[str, Any] | bytes | str | None = None)

 # #region ScenarioExecution.PublishedCatalogSource.Load [C:4] [TYPE Function] [SEMANTICS baseline,catalog,published,loader,fail-closed]
 # @ingroup ScenarioExecution
-# @BRIEF Fetch one published catalog generation for the explicit selector; every failure is None.
+# @BRIEF Fetch the exact Git catalog bytes for the explicit selector; every failure is None.
 # @PRE baseline_set and baseline_set_version are non-empty (the resolver consults the catalog only
 #      for an explicit selector; refs without a selector fail BASELINE_MISSING before any fetch).
-# @POST 200 with JSON object returns the parsed snapshot; unconfigured source, missing selector,
-#       transport error, non-200, or non-object payload return None without raising.
+#      ref_override, when supplied by the receipt binder, is an observed lowercase 40-hex Git head.
+# @POST 200 returns exact response bytes; unconfigured source, missing selector, invalid ref_override,
+#       transport error or non-200 returns None without raising.
 # @SIDE_EFFECT One bounded outbound HTTP GET to the configured Gitea raw-content endpoint.
 # @REJECTED Caching the fetched snapshot was rejected — a cached generation could outlive a
 #           superseded publication and resolve a stale pin; each start fetches the current bytes.
-def load_published_catalog_from_config(
+def load_published_catalog_bytes_from_config(
    baseline_set: str | None,
    baseline_set_version: str | None,
    *,
    transport: httpx.BaseTransport | None = None,
-) -> dict[str, Any] | None:
+    ref_override: str | None = None,
+) -> bytes | None:
    config = _source_config()
    if config is None:
        return None
@@ -110,6 +112,8 @@ def load_published_catalog_from_config(
    set_version = (baseline_set_version or "").strip()
    if not set_id or not set_version:
        return None
+    if ref_override is not None and (len(ref_override) != 40 or any(char not in "0123456789abcdef" for char in ref_override)):
+        return None
    try:
        # Accept both placeholder families so a single PUBLISHED_CATALOG_PATH_TEMPLATE works for the
        # publisher ({baseline_set_id}/{version}) and this loader ({baseline_set}/{baseline_set_version}).
@@ -130,7 +134,7 @@ def load_published_catalog_from_config(
            timeout=_CATALOG_SOURCE_TIMEOUT_SECONDS,
            headers={"Authorization": f"token {config['token']}"},
        ) as client:
-            response = client.get(url, params={"ref": config["ref"]})
+            response = client.get(url, params={"ref": ref_override or config["ref"]})
    except Exception as exc:
        logger.explore(
            "Published catalog source unreachable", src=_SRC, error=repr(exc),
@@ -143,19 +147,40 @@ def load_published_catalog_from_config(
            payload={"status_code": response.status_code, "repo": config["repo"], "ref": config["ref"]},
        )
        return None
+    return response.content
+# #endregion ScenarioExecution.PublishedCatalogSource.Load
+
+
+# #region ScenarioExecution.PublishedCatalogSource.Parse [C:2] [TYPE Function] [SEMANTICS baseline,catalog,published,loader,parse]
+# @BRIEF Parse the Git catalog bytes for existing callers; the receipt binder uses exact bytes.
+# @POST Returns a JSON object or None for a missing, malformed or non-object response.
+def load_published_catalog_from_config(
+    baseline_set: str | None,
+    baseline_set_version: str | None,
+    *,
+    transport: httpx.BaseTransport | None = None,
+    ref_override: str | None = None,
+) -> dict[str, Any] | None:
+    raw = load_published_catalog_bytes_from_config(
+        baseline_set, baseline_set_version, transport=transport, ref_override=ref_override,
+    )
+    if raw is None:
+        return None
    try:
-        payload = response.json()
-    except ValueError:
+        payload = json.loads(raw)
+    except (ValueError, UnicodeDecodeError):
        logger.explore("Published catalog bytes are not JSON", src=_SRC)
        return None
    if not isinstance(payload, dict):
        logger.explore("Published catalog payload is not an object", src=_SRC)
        return None
+    config = _source_config()
    logger.reflect(
        "Published catalog snapshot loaded", src=_SRC,
-        payload={"repo": config["repo"], "ref": config["ref"], "baseline_set": set_id, "baseline_set_version": set_version},
+        payload={"repo": config["repo"] if config else "", "ref": config["ref"] if config else "",
+                 "baseline_set": baseline_set, "baseline_set_version": baseline_set_version},
    )
    return payload
-# #endregion ScenarioExecution.PublishedCatalogSource.Load
+# #endregion ScenarioExecution.PublishedCatalogSource.Parse

 # #endregion ScenarioExecution.PublishedCatalogSource
--- a/backend/src/services/dashboard_testing/execution/start_run.py
+++ b/backend/src/services/dashboard_testing/execution/start_run.py
@@ -51,7 +51,7 @@ from .prod_guards import (
    reject_prod_browser_mutation as _reject_prod_browser_mutation,
    reject_unverified_prod_context as _reject_unverified_prod_context,
 )
-from .published_catalog_source import load_published_catalog_from_config
+from .publication_receipt_binding import load_receipted_published_catalog
 from .runner_plan import derive_runner_plan, resolve_pinned_policy
 from .start_contract import classify_start_error, compute_request_hash

@@ -239,7 +239,7 @@ def start_run(
    _reject_automated_human_plan(plan, trigger_source)
    catalog_snapshot = load_published_catalog(published_catalog)
    if catalog_snapshot is None and baseline_set and baseline_set_version:
-        catalog_snapshot = load_published_catalog_from_config(baseline_set, baseline_set_version)
+        catalog_snapshot = load_receipted_published_catalog(db, baseline_set, baseline_set_version)
    try:
        baseline_pin = resolve_baseline_pin(
            graph_or_plan=plan,
--- a/backend/tests/services/dashboard_testing/execution/test_publication_git_repository.py
+++ b/backend/tests/services/dashboard_testing/execution/test_publication_git_repository.py
@@ -0,0 +1,200 @@
+# #region Test.ScenarioExecution.PublicationGitRepository [C:3] [TYPE Module] [SEMANTICS test,publish,git,receipt,revision]
+# @BRIEF Exercise the publication worker against a Gitea API adapter backed by an actual isolated Git repository.
+# @RELATION VERIFIES -> [ScenarioExecution.PublicationWorker]
+# @TEST_INVARIANT A published receipt names the Git commit containing the exact envelope bytes, and
+#                 the same operation settles the bound revision without a second commit on retry.
+import base64
+from copy import deepcopy
+import json
+import subprocess
+from pathlib import Path
+
+import httpx
+
+from src.models.publication_operation import PublicationOperation
+from src.services.dashboard_testing.catalog_revision_log import append_revision, current_head, read_revisions
+from src.services.dashboard_testing.execution.publication_worker import publish_baseline_catalog
+from src.services.dashboard_testing.execution.publication_receipt_binding import load_receipted_published_catalog
+from src.services.dashboard_testing.execution.baseline_resolver import resolve_baseline_pin
+from src.services.dashboard_testing.filters import _compute_filters_hash
+from src.services.dashboard_testing.reference_source import build_reference_source
+from src.schemas.dashboard_testing.filters import NormalizedFilter
+
+
+def _git(repo: Path, *args: str, input_bytes: bytes | None = None) -> bytes:
+    return subprocess.run(
+        ["git", "-C", str(repo), *args], input=input_bytes, capture_output=True, check=True,
+    ).stdout.strip()
+
+
+def test_published_receipt_and_revision_name_real_git_commit(monkeypatch, db_session, tmp_path):
+    repo = tmp_path / "catalog-repo"
+    repo.mkdir()
+    _git(repo, "init", "-q", "-b", "main")
+    _git(repo, "config", "user.name", "Publication Test")
+    _git(repo, "config", "user.email", "publication-test@example.invalid")
+    (repo / "README").write_text("isolated publication repository\n", encoding="utf-8")
+    _git(repo, "add", "README")
+    _git(repo, "commit", "-qm", "Initial commit")
+    initial_head = _git(repo, "rev-parse", "HEAD").decode()
+
+    catalog = tmp_path / "local" / "baselines.yaml"
+    platform_filter = NormalizedFilter(
+        filter_id="platform", dataset_id=20, column="Platform", operator="IN", value={"values": ["3DS"]},
+        target_chart_ids=[96],
+    )
+    filters_hash = _compute_filters_hash([platform_filter])
+    source = build_reference_source(
+        url="https://example.invalid/superset/dashboard/10/?native_filters_key=test-key",
+        url_kind="native_filters_key",
+        source_url_ref="draft:test-run:reference-url", environment_id="test-environment",
+        dashboard_id=10, filters={"filters": [platform_filter.model_dump(mode="json")], "filters_hash": filters_hash},
+        query_model_fingerprint=f"sha256:{'d' * 64}",
+    )
+    entry = {
+        "baseline_id": "11111111-1111-4111-8111-111111111111",
+        "baseline_revision_id": "22222222-2222-4222-8222-222222222221",
+        "entry_digest": "a" * 64,
+        "coordinate_hash": "c" * 64,
+        "status": "approved",
+        "supersedes_revision_id": None,
+        "entry": {
+            "baseline_id": "11111111-1111-4111-8111-111111111111", "kind": "metric",
+            "dashboard_id": 10, "provenance": {"environment": "test-environment"},
+            "normalized_filters": {"filters_hash": filters_hash},
+            "release_version": "v1.0.0", "release_commit_hash": initial_head,
+            "source_response_hash": "e" * 64,
+        },
+        "reference_source": source,
+        "capture_artifact_id": "33333333-3333-4333-8333-333333333331",
+        "capture_profile_hash": "b" * 64,
+        "review_disposition_id": None,
+        "review_status": "not_applicable",
+    }
+    revision = append_revision(catalog, entry_revisions=[entry], actor_id="qa", reason="approved", expected_cas=0)
+    envelope = {
+        "baseline_set_id": "reference-url-test", "baseline_set_version": "1",
+        "release_id": "11111111-1111-4111-8111-111111111111",
+        "baseline_family": "a" * 64,
+        "catalog_digest": revision["catalog_digest"],
+        "catalog_revision": {**revision, "publication": {**revision["publication"], "state": "published"}},
+    }
+    for key, value in {
+        "PUBLISHED_CATALOG_GITEA_URL": "http://isolated-git.test",
+        "PUBLISHED_CATALOG_GITEA_TOKEN": "test-token",
+        "PUBLISHED_CATALOG_REPO": "isolated/catalog",
+        "PUBLISHED_CATALOG_REF": "main",
+    }.items():
+        monkeypatch.setenv(key, value)
+
+    target_path = "catalogs/reference-url-test/v1.json"
+    raw_override: bytes | None = None
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        path = request.url.path
+        if path.endswith("/branches/main") and request.method == "GET":
+            return httpx.Response(200, json={"commit": {"id": _git(repo, "rev-parse", "HEAD").decode()}})
+        if "/raw/" in path and request.method == "GET":
+            if raw_override is not None:
+                return httpx.Response(200, content=raw_override)
+            relative = path.split("/raw/", 1)[1]
+            requested_ref = request.url.params.get("ref")
+            try:
+                content = _git(repo, "show", f"{requested_ref}:{relative}")
+            except subprocess.CalledProcessError:
+                return httpx.Response(404)
+            return httpx.Response(200, content=content)
+        if "/contents/" in path:
+            relative = path.split("/contents/", 1)[1]
+            assert relative == target_path
+            destination = repo / relative
+            if request.method == "GET":
+                if not destination.exists():
+                    return httpx.Response(404)
+                return httpx.Response(200, json={"sha": _git(repo, "hash-object", relative).decode(),
+                                                  "content": base64.b64encode(destination.read_bytes()).decode()})
+            assert request.method == "POST"
+            body = json.loads(request.content)
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            destination.write_bytes(base64.b64decode(body["content"]))
+            _git(repo, "add", relative)
+            _git(repo, "commit", "-qm", body["message"])
+            return httpx.Response(201, json={"commit": {"id": _git(repo, "rev-parse", "HEAD").decode()}})
+        return httpx.Response(404)
+
+    request = {
+        "envelope": envelope, "expected_branch_head": initial_head,
+        "idempotency_key": "isolated-git-publication", "actor": "qa",
+        "transport": httpx.MockTransport(handler), "local_catalog_path": catalog,
+    }
+    result = publish_baseline_catalog(db_session, **request)
+    commit = _git(repo, "rev-parse", "HEAD").decode()
+    assert commit != initial_head
+    assert result["state"] == "published"
+    assert result["commit_sha"] == result["published_receipt"]["head_after"] == commit
+    assert result["published_receipt"]["head_before"] == initial_head
+    assert json.loads(_git(repo, "show", f"HEAD:{target_path}")) == envelope
+    assert json.loads(_git(repo, "show", f"HEAD:{target_path}"))["catalog_revision"]["entry_revisions"][0]["reference_source"]["resolved_native_filters"][0]["value"]["values"] == ["3DS"]
+    assert current_head(catalog)["publication"]["state"] == "published"
+    assert current_head(catalog)["publication"]["commit_hash"] == commit
+    assert current_head(catalog)["publication"]["published_receipt_id"] == result["operation_id"]
+    assert len(read_revisions(catalog)) == 1
+    assert db_session.query(PublicationOperation).one().commit_sha == commit
+
+    loaded = load_receipted_published_catalog(
+        db_session, "reference-url-test", "1", transport=request["transport"],
+    )
+    assert loaded is not None
+    assert loaded["catalog_revision"]["publication"]["commit_hash"] == commit
+    assert loaded["catalog_revision"]["publication"]["published_receipt_id"] == result["operation_id"]
+    pin = resolve_baseline_pin(
+        graph_or_plan={"baselines": {"metric": {"reference": "reference-url-test"}}},
+        baseline_set="reference-url-test", baseline_set_version="1", published_catalog=loaded,
+        environment_id="test-environment", dashboard_id=10,
+    )
+    assert pin["publication_commit_hash"] == commit
+    assert pin["entries"][0]["reference_source"]["resolved_native_filters"][0]["value"]["values"] == ["3DS"]
+
+    raw_override = json.dumps(envelope, sort_keys=True, ensure_ascii=False, indent=2).encode()
+    assert load_receipted_published_catalog(
+        db_session, "reference-url-test", "1", transport=request["transport"],
+    ) is None  # Identical parsed JSON is not the committed publication bytes.
+    raw_override = None
+
+    operation = db_session.query(PublicationOperation).one()
+    original_envelope = operation.envelope
+    original_digest = operation.catalog_digest
+    original_receipt = operation.published_receipt
+    corrupt = deepcopy(envelope)
+    corrupt["catalog_digest"] = "f" * 64
+    corrupt["catalog_revision"]["catalog_digest"] = "f" * 64
+    operation.envelope = corrupt
+    operation.catalog_digest = "f" * 64
+    db_session.commit()
+    raw_override = json.dumps(corrupt, sort_keys=True, ensure_ascii=False).encode()
+    assert load_receipted_published_catalog(
+        db_session, "reference-url-test", "1", transport=request["transport"],
+    ) is None  # Both declared digests agree, but neither hashes entry_revisions.
+    raw_override = None
+    operation.envelope = original_envelope
+    operation.catalog_digest = original_digest
+    db_session.commit()
+
+    operation.published_receipt = {**original_receipt, "head_after": initial_head}
+    db_session.commit()
+    assert load_receipted_published_catalog(
+        db_session, "reference-url-test", "1", transport=request["transport"],
+    ) is None
+    operation.published_receipt = original_receipt
+    db_session.commit()
+
+    assert publish_baseline_catalog(db_session, **request)["operation_id"] == result["operation_id"]
+    assert _git(repo, "rev-list", "--count", "HEAD").decode() == "2"
+
+    (repo / "README").write_text("branch moved after publication\n", encoding="utf-8")
+    _git(repo, "add", "README")
+    _git(repo, "commit", "-qm", "Unrelated branch move")
+    assert load_receipted_published_catalog(
+        db_session, "reference-url-test", "1", transport=request["transport"],
+    ) is None
+# #endregion Test.ScenarioExecution.PublicationGitRepository
--- a/backend/tests/services/dashboard_testing/execution/test_publication_worker.py
+++ b/backend/tests/services/dashboard_testing/execution/test_publication_worker.py
@@ -41,7 +41,7 @@ _ENVELOPE = {
    "baseline_set_id": "ss-prod-visual", "baseline_set_version": "1",
    "release_id": "11111111-1111-4111-8111-111111111111", "baseline_family": "a" * 64,
    "catalog_digest": "a" * 64,
-    "catalog_revision": {"catalog_revision_id": "rev-001", "publication": {"state": "published"}},
+    "catalog_revision": {"catalog_revision_id": "rev-001", "entry_revisions": [], "publication": {"state": "published"}},
 }
 _HEAD_A = "b" * 40
 _HEAD_B = "c" * 40
--- a/backend/tests/services/dashboard_testing/execution/test_start_run_period_receipt.py
+++ b/backend/tests/services/dashboard_testing/execution/test_start_run_period_receipt.py
@@ -90,7 +90,7 @@ def test_rest_period_block_writes_exactly_one_receipt(period_parity, monkeypatch
        lambda injected=None: _closed_period_snapshot(),
    )
    monkeypatch.setattr(
-        "src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
+        "src.services.dashboard_testing.execution.start_run.load_receipted_published_catalog",
        lambda *_args, **_kwargs: None,
    )
    _patch_config(monkeypatch)
@@ -127,7 +127,7 @@ def test_mcp_period_block_writes_exactly_one_receipt(period_parity, monkeypatch)
        lambda injected=None: _closed_period_snapshot(),
    )
    monkeypatch.setattr(
-        "src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
+        "src.services.dashboard_testing.execution.start_run.load_receipted_published_catalog",
        lambda *_args, **_kwargs: None,
    )
    _patch_config(monkeypatch)
@@ -162,7 +162,7 @@ def test_shared_launch_identity_collapses_to_one_receipt(period_parity, monkeypa
        lambda injected=None: _closed_period_snapshot(),
    )
    monkeypatch.setattr(
-        "src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
+        "src.services.dashboard_testing.execution.start_run.load_receipted_published_catalog",
        lambda *_args, **_kwargs: None,
    )
    _patch_config(monkeypatch)
@@ -202,7 +202,7 @@ def test_digest_mismatch_writes_no_receipt(period_parity, monkeypatch) -> None:
        lambda injected=None: _catalog_snapshot(revision_digest="not-a-sha256"),
    )
    monkeypatch.setattr(
-        "src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
+        "src.services.dashboard_testing.execution.start_run.load_receipted_published_catalog",
        lambda *_args, **_kwargs: None,
    )
    _patch_config(monkeypatch)
@@ -234,7 +234,7 @@ def test_period_block_creates_no_run_row(period_parity, monkeypatch) -> None:
        lambda injected=None: _closed_period_snapshot(),
    )
    monkeypatch.setattr(
-        "src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
+        "src.services.dashboard_testing.execution.start_run.load_receipted_published_catalog",
        lambda *_args, **_kwargs: None,
    )
    _patch_config(monkeypatch)
@@ -263,7 +263,7 @@ def test_receipt_emission_failure_never_masks_the_block(period_parity, monkeypat
        lambda injected=None: _closed_period_snapshot(),
    )
    monkeypatch.setattr(
-        "src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
+        "src.services.dashboard_testing.execution.start_run.load_receipted_published_catalog",
        lambda *_args, **_kwargs: None,
    )
    _patch_config(monkeypatch)
--- a/backend/tests/services/dashboard_testing/registry/test_scheduled_scenario_start.py
+++ b/backend/tests/services/dashboard_testing/registry/test_scheduled_scenario_start.py
@@ -39,11 +39,13 @@
 from __future__ import annotations

 import hashlib
+import json
 from datetime import UTC, datetime
+from pathlib import Path
 from unittest.mock import MagicMock

 from src.models.scenario_approval import ActionApprovalGate
-from src.models.scenario_automation import ScenarioNotificationEvent
+from src.models.scenario_automation import ScenarioNotificationEvent, ScenarioSchedule
 from src.models.scenario_registry import ScenarioRegistryEntry, ScenarioRevision
 from src.models.scenario_run import ScenarioRun
 from src.services.dashboard_testing.scenario.templates import (
@@ -93,13 +95,14 @@ def _human_graph() -> dict:
 # #region Test.ScenarioExecution.ScheduledStart.PersistFixture [C:2] [TYPE Function] [SEMANTICS test,fixture,registry,seed]
 # @ingroup Test.ScenarioExecution.ScheduledStart
 # @BRIEF Persist one registry entry and its revisions; no executor or lifecycle boundary is mocked.
-def _persist_scenario(db, *, current_revision_id: str, graph_by_revision: dict[str, dict]) -> None:
+def _persist_scenario(db, *, current_revision_id: str, graph_by_revision: dict[str, dict],
+                      environment_id: str = "prod") -> None:
    db.add(ScenarioRegistryEntry(
        scenario_id=_SCENARIO_ID,
        scenario_key="scheduled-start-046",
        name="Scheduled start 046",
        dashboard_id=46,
-        environment_ids=["prod"],
+        environment_ids=[environment_id],
        owner_id="analyst-046",
        owner_username="analyst.046",
        lifecycle_status="READY",
@@ -298,6 +301,112 @@ def test_scheduled_callback_stamps_baseline_pin_in_due_identity(
 # #endregion Test.ScenarioExecution.ScheduledStart.BaselineStamp


+# #region Test.ScenarioExecution.ScheduledStart.ReferencePinAdmission [C:2] [TYPE Function] [SEMANTICS test,scheduled,baseline,reference,pin]
+# @ingroup Test.ScenarioExecution.ScheduledStart
+# @BRIEF A registered APScheduler job admits and replays a frozen URL-backed pin with the real resolver.
+# @TEST_INVARIANT Core.Scheduler.ExecuteScheduledScenario: a scheduled run with a baseline reference
+#                  carries the published URL/filter pin before dispatch and a same-second replay
+#                  preserves the request hash without consulting the expired short key. -> VERIFIED_BY:
+#                  test_scheduled_baseline_reference_reaches_real_resolver
+def test_scheduled_baseline_reference_reaches_real_resolver(
+    registry_session, monkeypatch, published_baseline_pin_for_registry_starts,
+):
+    from src.services.dashboard_testing.execution import baseline_resolver, start_run as start_module
+    from src.schemas.dashboard_testing.filters import NormalizedFilter
+    from src.services.dashboard_testing.filters import _compute_filters_hash
+    from src.services.dashboard_testing.reference_source import build_reference_source
+
+    refresh_path = Path(__file__).resolve().parents[5] / "specs/044-dashboard-scenario-execution/fixtures/production-contract-refresh.json"
+    refresh = json.loads(refresh_path.read_text(encoding="utf-8"))
+    platform = NormalizedFilter(filter_id="platform", dataset_id=20, column="Platform", operator="IN", value={"values": ["3DS"]}, target_chart_ids=[96])
+    filters_hash = _compute_filters_hash([platform])
+    source = build_reference_source(
+        url="https://example.invalid/superset/dashboard/46/?native_filters_key=expired-key",
+        url_kind="native_filters_key", source_url_ref="draft:scheduled-test:reference-url",
+        environment_id="preprod", dashboard_id=46,
+        filters={"filters": [platform.model_dump(mode="json")], "filters_hash": filters_hash},
+        query_model_fingerprint=f"sha256:{'d' * 64}",
+    )
+    revision = refresh["catalog_revision"]
+    metric = revision["entry_revisions"][1]
+    revision["entry_revisions"] = [metric]
+    metric["reference_source"] = source
+    metric["entry"]["dashboard_id"] = 46
+    metric["entry"]["provenance"]["environment"] = "preprod"
+    metric["entry"]["normalized_filters"] = {"schema_version": 1, "filters": [platform.model_dump(mode="json")], "filters_hash": filters_hash}
+    snapshot = {key: refresh["baseline_pin"][key] for key in ("baseline_set_id", "baseline_set_version", "release_id", "baseline_family", "catalog_digest")}
+    snapshot["catalog_revision"] = revision
+    monkeypatch.setattr(start_module, "load_published_catalog", lambda injected=None: injected if injected is not None else snapshot)
+    from src.core import scheduler as scheduler_module
+
+    class _FrozenDatetime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return datetime(2026, 9, 25, 12, 0, 5, tzinfo=tz or UTC)
+
+    monkeypatch.setattr(scheduler_module, "datetime", _FrozenDatetime)
+
+    graph = _assertion_graph()
+    graph["baselines"] = {"revenue_sum": {"reference": "ss-prod-visual"}}
+    _persist_scenario(
+        registry_session,
+        current_revision_id=_CURRENT_REVISION_ID,
+        graph_by_revision={_CURRENT_REVISION_ID: graph},
+        environment_id="preprod",
+    )
+    expected_pin = published_baseline_pin_for_registry_starts
+    registry_session.add(ScenarioSchedule(
+        id=_SCHEDULE_ID, scenario_id=_SCENARIO_ID, revision_policy="current",
+        environment_id="preprod", cron_expr="* * * * *", timezone="UTC",
+        baseline_set=expected_pin["baseline_set_id"],
+        baseline_set_version=expected_pin["baseline_set_version"],
+        created_by="analyst-046",
+    ))
+    registry_session.flush()
+    # The registry test fixture substitutes the resolver; restore production admission here.
+    monkeypatch.setattr(start_module, "resolve_baseline_pin", baseline_resolver.resolve_baseline_pin)
+    from src.services.dashboard_testing import reference_discovery
+
+    resolve_expired_key = MagicMock(side_effect=AssertionError("expired native_filters_key was re-resolved"))
+    monkeypatch.setattr(reference_discovery, "_key_state", resolve_expired_key)
+
+    service = object.__new__(scheduler_module.SchedulerService)
+    service.scheduler = MagicMock()
+    service.add_scenario_job(
+        schedule_id=_SCHEDULE_ID, scenario_id=_SCENARIO_ID, revision_policy="current",
+        revision_id=None, environment_id="preprod", cron_expr="* * * * *", timezone="UTC",
+    )
+    registered = service.scheduler.add_job.call_args
+    from sqlalchemy.orm import sessionmaker
+
+    scheduler_session = sessionmaker(bind=registry_session.bind)
+    registry_session.commit()
+    monkeypatch.setattr(scheduler_module, "SessionLocal", scheduler_session)
+    fire_registered_job = lambda: registered.args[0](*registered.kwargs["args"])
+    fire_registered_job()
+
+    run = registry_session.query(ScenarioRun).filter(ScenarioRun.scenario_id == _SCENARIO_ID).one()
+    assert run.trigger_source == "scheduled"
+    pin = (run.runner_plan or {})["baseline_pin"]
+    assert pin["baseline_set_id"] == "ss-prod-visual"
+    assert pin["entries"][0]["reference_source"] == source
+    assert (run.target_snapshot or {})["baseline_set"] == expected_pin["baseline_set_id"]
+    assert (run.target_snapshot or {})["baseline_set_version"] == expected_pin["baseline_set_version"]
+    original_hash = run.request_hash
+    assert run.status == "queued"
+    scheduler_module.execute_scheduled_queued_scenario_dispatch()
+    registry_session.expire_all()
+    run = registry_session.get(ScenarioRun, run.id)
+    assert run.status == "passed"
+    assert (run.runner_plan or {})["baseline_pin"] == pin
+    assert (run.target_snapshot or {})["baseline_pin"] == pin
+    fire_registered_job()
+    assert registry_session.query(ScenarioRun).filter(ScenarioRun.scenario_id == _SCENARIO_ID).count() == 1
+    assert registry_session.query(ScenarioRun).filter(ScenarioRun.scenario_id == _SCENARIO_ID).one().request_hash == original_hash
+    resolve_expired_key.assert_not_called()
+# #endregion Test.ScenarioExecution.ScheduledStart.ReferencePinAdmission
+
+
 # #region Test.ScenarioExecution.ScheduledStart.BaselineConflict [C:4] [TYPE Function] [SEMANTICS test,scheduled,baseline,pin,idempotency,conflict]
 # @ingroup Test.ScenarioExecution.ScheduledStart
 # @BRIEF Same-second due replay with a changed resolved baseline rejects typed; the pinned run is never silently reused.
--- a/backend/tests/services/dashboard_testing/test_reference_discovery.py
+++ b/backend/tests/services/dashboard_testing/test_reference_discovery.py
@@ -12,6 +12,7 @@ import pytest

 from src.core.superset_client._chart_data import ChartDataResponse
 from src.services.dashboard_testing.reference_discovery import discover_reference_dashboard
+from src.services.dashboard_testing.reference_selection import capture_reference_selection


 # #region Test.ReferenceDiscovery.ClientFixture [C:3] [TYPE Function]
@@ -337,4 +338,88 @@ async def test_permalink_uses_authoritative_dashboard_uuid():
            client, "preprod", 42, "https://superset.example/superset/dashboard/p/abc123/",
        )
 # #endregion Test.ReferenceDiscovery.PermalinkUuid
+
+
+# #region Test.ReferenceDiscovery.SuppliedPermalink [C:2] [TYPE Function]
+# @TEST_INVARIANT The supplied permalink path resolves its dashboard and filter state through the configured client.
+@pytest.mark.asyncio
+async def test_supplied_permalink_path_resolves_authoritative_state():
+    client = _client()
+    preview = await discover_reference_dashboard(
+        client, "preprod", 42,
+        "https://superset.example/superset/dashboard/p/9Q7ZpedqO3d/",
+        max_queries=1,
+    )
+    client.get_dashboard_permalink_state.assert_awaited_once_with("9Q7ZpedqO3d")
+    assert preview["filters"]["filters"][0]["value"]["values"] == ["East"]
+    assert preview["coordinates"][0]["execution_status"] == "executed"
+# #endregion Test.ReferenceDiscovery.SuppliedPermalink
+
+
+# #region Test.ReferenceDiscovery.TypedVectors [C:2] [TYPE Function]
+# @TEST_INVARIANT Date and numeric URL states reach chart-data with their Superset operators and hardcoded values.
+@pytest.mark.asyncio
+@pytest.mark.parametrize("filter_type,filter_id,column,entry,expected", [
+    ("filter_date", "period", "sale_date",
+     {"extraFormData": {"time_range": "2024-01-01 : 2024-02-01"}},
+     {"operator": "TEMPORAL_RANGE", "subject": "sale_date", "from": "2024-01-01", "to": "2024-02-01"}),
+    ("filter_range", "price", "price",
+     {"extraFormData": {"filters": [{"col": "price", "op": ">=", "val": 10},
+                                    {"col": "price", "op": "<=", "val": 20}]}},
+     {"operator": ">=", "subject": "price", "comparator": "10"}),
+])
+async def test_time_and_numeric_filter_vectors_reach_chart_data(filter_type, filter_id, column, entry, expected):
+    client = _client()
+    dashboard = client.get_dashboard.return_value["result"]
+    metadata = json.loads(dashboard["json_metadata"])
+    metadata["native_filter_configuration"] = [{
+        "id": filter_id, "name": filter_id, "filterType": filter_type,
+        "targets": [{"datasetId": 7, "column": {"name": column}}],
+        "scope": {"rootPath": ["ROOT"], "excluded": [12]},
+    }]
+    dashboard["json_metadata"] = json.dumps(metadata)
+    client.get_native_filter_state.return_value = {"result": {"value": json.dumps({filter_id: entry})}}
+    preview = await discover_reference_dashboard(
+        client, "preprod", 42,
+        "https://superset.example/superset/dashboard/42/?native_filters_key=key123",
+        max_queries=1,
+    )
+    assert preview["coordinates"][0]["execution_status"] == "executed"
+    filters = client.execute_chart_data_raw.await_args.kwargs["filters"]
+    assert filters[0].items() >= expected.items()
+    if filter_type == "filter_range":
+        assert len(filters) == 2
+        assert filters[1].items() >= {"operator": "<=", "subject": "price", "comparator": "20"}.items()
+# #endregion Test.ReferenceDiscovery.TypedVectors
+
+
+# #region Test.ReferenceDiscovery.ChangedKey [C:2] [TYPE Function]
+# @TEST_INVARIANT A key whose active value changes between preview and capture creates no run.
+@pytest.mark.asyncio
+async def test_changed_key_state_aborts_capture_before_write():
+    client = _client()
+    url = "https://superset.example/superset/dashboard/42/?native_filters_key=key123"
+    original = await discover_reference_dashboard(client, "preprod", 42, url, max_queries=1)
+    client.get_native_filter_state.return_value = {"result": {"value": json.dumps({
+        "region": {"filterState": {"value": ["West"]}},
+    })}}
+    selected = SimpleNamespace(
+        chart_id=11, dataset_id=7, metric="sum_sales", execution_status="executed",
+        evidence_digest=original["coordinates"][0]["evidence_digest"],
+        value_digest=original["coordinates"][0]["value_digest"],
+    )
+    class RecordingDB:
+        added = False
+
+        def add(self, _value):
+            self.added = True
+
+    db = RecordingDB()
+    with pytest.raises(ValueError, match="Native filter state changed"):
+        await capture_reference_selection(
+            db, "user-1", client, "preprod", 42, url, [selected],
+            original["query_model_fingerprint"], original["filters"]["filters_hash"],
+        )
+    assert db.added is False
+# #endregion Test.ReferenceDiscovery.ChangedKey
 # #endregion Test.ReferenceDiscovery
--- a/backend/tests/test_mcp_client_flow_http.py
+++ b/backend/tests/test_mcp_client_flow_http.py
@@ -13,8 +13,10 @@ from __future__ import annotations

 import base64
 import hashlib
+import json
 import os
 import secrets
+from unittest.mock import AsyncMock
 from urllib.parse import parse_qs, urlparse
 from uuid import uuid4

@@ -30,6 +32,7 @@ from src.app import app
 from src.core.auth.security import get_password_hash
 from src.core.database import SessionLocal
 from src.mcp_server.server import create_mcp_asgi_app
+import src.mcp_server.tools_review as review_module
 from src.models.auth import (
    McpToolInvocationRecord,
    OAuthAuthorizationCode,
@@ -266,4 +269,97 @@ def test_scripted_user_client_pkce_http_flow(registrant, mcp_signing_key) -> Non
        assert '"decide_approval"' not in listed.text
 # #endregion Test.McpClientFlowHttp.PkceFlow

+
+# #region Test.McpClientFlowHttp.ReferenceHuman [C:4] [TYPE Function]
+# @TEST_INVARIANT Human OAuth identity uses live DB grants to expose preview and capture over MCP HTTP.
+# @TEST_EDGE missing_write -> capture stays hidden until the human role receives WRITE.
+def test_reference_tools_human_oauth_http_rbac(registrant, mcp_signing_key, monkeypatch) -> None:
+    with SessionLocal.begin() as db:
+        user = db.query(User).filter(User.username == registrant["username"]).one()
+        role = user.roles[0]
+        role.permissions.extend([
+            Permission(resource="dashboard:testing", action="READ"),
+            Permission(resource="dashboard:testing", action="EXECUTE"),
+        ])
+    preview = AsyncMock(return_value={"dashboard_id": 42, "filters": {"filters": []}})
+    capture = AsyncMock(return_value={"candidate_ids": ["isolated-candidate"]})
+    monkeypatch.setattr(review_module, "reference_discovery", preview)
+    monkeypatch.setattr(review_module, "capture_reference_selection_endpoint", capture)
+
+    oauth = TestClient(app)
+    web_token = _login(registrant["username"], registrant["password"])
+    registered = oauth.post("/oauth/register", json={
+        "client_name": "reference-human-test", "redirect_uris": ["http://localhost:9999/cb"],
+        "scope": "mcp:read",
+    })
+    assert registered.status_code == 201
+    verifier = secrets.token_urlsafe(48)
+    challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode("ascii")).digest()).rstrip(b"=").decode("ascii")
+    approved = oauth.post("/api/oauth/authorize/approve", headers={"Authorization": f"Bearer {web_token}"}, json={
+        "client_id": registered.json()["client_id"], "redirect_uri": "http://localhost:9999/cb",
+        "code_challenge": challenge, "code_challenge_method": "S256", "response_type": "code",
+        "scope": "mcp:read", "state": "reference-test",
+    })
+    assert approved.status_code == 200
+    code = parse_qs(urlparse(approved.json()["redirect_url"]).query)["code"][0]
+    issued = oauth.post("/oauth/token", data={
+        "grant_type": "authorization_code", "client_id": registered.json()["client_id"],
+        "code": code, "redirect_uri": "http://localhost:9999/cb", "code_verifier": verifier,
+    })
+    assert issued.status_code == 200
+    headers = {"Authorization": f"Bearer {issued.json()['access_token']}",
+               "Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
+
+    def rpc(response):
+        assert response.status_code == 200
+        return next(json.loads(line[6:]) for line in response.text.splitlines() if line.startswith("data: "))
+
+    with TestClient(create_mcp_asgi_app(), base_url="http://testserver") as mcp:
+        initialized = mcp.post("/", headers=headers, json={
+            "jsonrpc": "2.0", "id": 1, "method": "initialize",
+            "params": {"protocolVersion": "2025-06-18", "capabilities": {},
+                       "clientInfo": {"name": "reference-human-test", "version": "1.0"}},
+        })
+        rpc(initialized)
+        session_headers = {**headers, "Mcp-Session-Id": initialized.headers["mcp-session-id"]}
+        assert mcp.post("/", headers=session_headers,
+                        json={"jsonrpc": "2.0", "method": "notifications/initialized"}).status_code == 202
+
+        def listed_names(request_id):
+            listed = rpc(mcp.post("/", headers=session_headers,
+                                  json={"jsonrpc": "2.0", "id": request_id, "method": "tools/list", "params": {}}))
+            return {tool["name"] for tool in listed["result"]["tools"]}
+
+        assert "preview_reference_dashboard" in listed_names(2)
+        assert "capture_reference_selection" not in listed_names(2)
+        with SessionLocal.begin() as db:
+            user = db.query(User).filter(User.username == registrant["username"]).one()
+            role = user.roles[0]
+            role.permissions.append(Permission(resource="dashboard:testing", action="WRITE"))
+        assert {"preview_reference_dashboard", "capture_reference_selection"} <= listed_names(3)
+
+        url = "https://superset.example/dashboard/42/?native_filters_key=fixture-key"
+        preview_result = rpc(mcp.post("/", headers=session_headers, json={
+            "jsonrpc": "2.0", "id": 4, "method": "tools/call",
+            "params": {"name": "preview_reference_dashboard", "arguments": {"request": {
+                "scenario_id": "isolated-scenario", "url": url, "max_queries": 1,
+            }}},
+        }))
+        assert preview_result["result"]["structuredContent"]["dashboard_id"] == 42
+        digest = "sha256:" + "a" * 64
+        capture_result = rpc(mcp.post("/", headers=session_headers, json={
+            "jsonrpc": "2.0", "id": 5, "method": "tools/call",
+            "params": {"name": "capture_reference_selection", "arguments": {"request": {
+                "scenario_id": "isolated-scenario", "url": url,
+                "selected": [{"chart_id": 11, "dataset_id": 7, "metric": "sum_sales",
+                              "evidence_digest": digest, "value_digest": digest,
+                              "execution_status": "executed"}],
+                "expected_query_model_fingerprint": digest, "expected_filters_hash": digest,
+            }}},
+        }))
+        assert capture_result["result"]["structuredContent"]["candidate_ids"] == ["isolated-candidate"]
+    assert preview.await_count == 1 and capture.await_count == 1
+    assert preview.await_args.kwargs["_reader"].id == capture.await_args.kwargs["_writer"].id
+# #endregion Test.McpClientFlowHttp.ReferenceHuman
+
 # #endregion Test.McpClientFlowHttp
--- a/frontend/src/lib/components/scenario-automation/AutomationPanel.svelte
+++ b/frontend/src/lib/components/scenario-automation/AutomationPanel.svelte
@@ -138,6 +138,8 @@
      scenario_id: schedule.scenario_id,
      environment_id: schedule.environment_id,
      cron_expr: schedule.cron_expr,
+      baseline_set: schedule.baseline_set,
+      baseline_set_version: schedule.baseline_set_version,
      revision_policy: schedule.revision_policy,
      revision_id: schedule.revision_id,
      timezone: schedule.timezone,
--- a/frontend/src/lib/components/scenario-automation/ScheduleForm.svelte
+++ b/frontend/src/lib/components/scenario-automation/ScheduleForm.svelte
@@ -62,6 +62,8 @@
  let timezone = $state("UTC");
  let revisionPolicy: RevisionPolicy = $state("current");
  let revisionId = $state("");
+  let baselineSet = $state("");
+  let baselineSetVersion = $state("");
  let policyId = $state("");
  let missedPolicy: MissedExecutionPolicy = $state("skip");
  let enabled = $state(true);
@@ -150,6 +152,8 @@
      timezone = editing.timezone;
      revisionPolicy = editing.revision_policy;
      revisionId = editing.revision_id ?? "";
+      baselineSet = editing.baseline_set ?? "";
+      baselineSetVersion = editing.baseline_set_version ?? "";
      policyId = editing.policy_id ?? "";
      missedPolicy = editing.missed_execution_policy;
      enabled = editing.enabled;
@@ -161,6 +165,8 @@
      timezone = "UTC";
      revisionPolicy = "current";
      revisionId = "";
+      baselineSet = "";
+      baselineSetVersion = "";
      policyId = "";
      missedPolicy = "skip";
      enabled = true;
@@ -194,6 +200,10 @@
      error = "Select a scenario with an active revision or enter a revision ID in Advanced";
      return;
    }
+    if (Boolean(baselineSet.trim()) !== Boolean(baselineSetVersion.trim())) {
+      error = "Baseline set and version must be provided together";
+      return;
+    }
    if (previewError) {
      error = previewError;
      return;
@@ -209,6 +219,8 @@
          revisionPolicy === "pinned"
            ? (revisionId.trim() || selectedScenario?.current_revision || null)
            : null,
+        baseline_set: baselineSet.trim() || null,
+        baseline_set_version: baselineSetVersion.trim() || null,
        policy_id: policyId.trim() || null,
        missed_execution_policy: missedPolicy,
        enabled,
@@ -217,6 +229,8 @@
        scenarioId = "";
        environmentId = "";
        revisionId = "";
+        baselineSet = "";
+        baselineSetVersion = "";
        policyId = "";
      }
    } catch (e: unknown) {
@@ -296,6 +310,16 @@
      </label>
    </div>
  {/if}
+  <div class="grid gap-3 sm:grid-cols-2">
+    <label class="flex flex-col gap-1 text-sm text-text">
+      Published baseline set
+      <input bind:value={baselineSet} class="input input-sm" aria-label="Published baseline set" />
+    </label>
+    <label class="flex flex-col gap-1 text-sm text-text">
+      Published baseline version
+      <input bind:value={baselineSetVersion} class="input input-sm" aria-label="Published baseline version" />
+    </label>
+  </div>

  <div class="grid gap-3 sm:grid-cols-3">
    <label class="flex flex-col gap-1 text-sm text-text">
--- a/frontend/src/lib/components/scenario-automation/tests/ScheduleForm.test.ts
+++ b/frontend/src/lib/components/scenario-automation/tests/ScheduleForm.test.ts
@@ -59,12 +59,28 @@ describe("ScheduleForm", () => {
      timezone: "UTC",
      revision_policy: "current",
      revision_id: null,
+      baseline_set: null,
+      baseline_set_version: null,
      policy_id: null,
      missed_execution_policy: "skip",
      enabled: true,
    });
  });

+  it("submits the explicit published baseline selector", async () => {
+    const onSave = vi.fn().mockResolvedValue(undefined);
+    mount(onSave);
+    await fireEvent.change(screen.getByLabelText("Scenario"), { target: { value: "sc-1" } });
+    await fireEvent.change(screen.getByLabelText("Environment"), { target: { value: "env-preprod-01" } });
+    await fireEvent.input(screen.getByLabelText("Published baseline set"), { target: { value: "reference-url-test" } });
+    await fireEvent.input(screen.getByLabelText("Published baseline version"), { target: { value: "1" } });
+    await fireEvent.click(screen.getByLabelText("Create schedule"));
+    expect(onSave).toHaveBeenCalledWith(expect.objectContaining({
+      baseline_set: "reference-url-test",
+      baseline_set_version: "1",
+    }));
+  });
+
  it("pins the current server revision only when the analyst selects pinned", async () => {
    const onSave = vi.fn().mockResolvedValue(undefined);
    mount(onSave);
--- a/frontend/src/lib/components/scenario-editor/tests/BaselineReview.test.ts
+++ b/frontend/src/lib/components/scenario-editor/tests/BaselineReview.test.ts
@@ -59,7 +59,7 @@ describe("BaselineReview", () => {
    vi.mocked(requestApi).mockRejectedValue(Object.assign(new Error("Gate already decided"), { status: 409 }));
    render(BaselineReview, { scenarioId: "scenario-1", canApprove: true });
    expect(await screen.findByText("Captured revenue")).toBeTruthy();
-    const approve = screen.getByRole("button", { name: /Утвердить и опубликовать|Approve and publish/ });
+    const approve = screen.getByRole("button", { name: /Утвердить и сохранить baseline|Approve and save baseline/ });
    const reason = screen.getByRole("textbox", { name: /Причина решения|Review reason/ }) as HTMLTextAreaElement;
    expect(approve.hasAttribute("disabled")).toBe(true);
    await fireEvent.input(reason, { target: { value: "Checked source evidence" } });
@@ -72,5 +72,15 @@ describe("BaselineReview", () => {
    expect(reason.value).toBe("Checked source evidence");
    expect(vi.mocked(fetchApi).mock.calls.length).toBeGreaterThanOrEqual(2);
  });
+
+  it("describes consumed approval as local catalog state", async () => {
+    vi.mocked(fetchApi).mockImplementation(async (path) =>
+      path.includes("baseline-overview")
+        ? { state: "ready", release_version: "v1.0.0", environment_name: "Preprod", entries: [], total: 0 }
+        : [{ ...candidate, status: "consumed", gate_status: "consumed" }]);
+    render(BaselineReview, { scenarioId: "approved-scenario", canApprove: true });
+    expect(await screen.findByText(/Git publication is a separate step|Публикация в Git выполняется отдельно/)).toBeTruthy();
+    expect(screen.queryByRole("button", { name: /Approve and publish|Утвердить и опубликовать/ })).toBeNull();
+  });
 });
 // #endregion Test.ScenarioEditor.BaselineReview
--- a/frontend/src/lib/i18n/locales/en/dashboard-testing.json
+++ b/frontend/src/lib/i18n/locales/en/dashboard-testing.json
@@ -523,12 +523,12 @@
  "baseline_review_loading": "Loading candidates…",
  "baseline_review_error": "Could not load baseline candidates.",
  "baseline_review_recovery": "Reload to review the current state.",
-  "baseline_review_missing": "There are no new candidates to review. Published values are shown above; capture a candidate from a scenario run to make a new decision.",
+  "baseline_review_missing": "There are no new candidates to review. Approved catalog values are shown above; capture a candidate from a scenario run to make a new decision.",
  "baseline_overview_title": "Approved values",
  "baseline_overview_no_release": "This dashboard has no approved release. Approve a release, then capture a baseline candidate.",
-  "baseline_overview_missing": "An approved release exists, but no baseline has been published for it.",
-  "baseline_overview_ready": "Published values are available for this release.",
-  "baseline_overview_stale": "Some published values belong to another release and need review.",
+  "baseline_overview_missing": "An approved release exists, but no baseline has been approved for it.",
+  "baseline_overview_ready": "Approved catalog values are available for this release.",
+  "baseline_overview_stale": "Some approved catalog values belong to another release and need review.",
  "baseline_overview_entry_stale": "This value belongs to another release",
  "baseline_overview_more": "More entries",
  "baseline_review_status": "Status",
@@ -536,14 +536,14 @@
  "baseline_review_state_pending": "Awaiting review",
  "baseline_review_state_confirmed": "Decision recorded",
  "baseline_review_state_approved": "Decision recorded",
-  "baseline_review_state_consumed": "Published",
+  "baseline_review_state_consumed": "Approved in catalog",
  "baseline_review_state_denied": "Rejected",
  "baseline_review_state_unknown": "Status needs review",
  "baseline_review_action_draft": "Check the source, value and comparison rule, then enter a reason for your decision.",
  "baseline_review_action_pending": "Check the source, value and comparison rule, then enter a reason for your decision.",
-  "baseline_review_action_confirmed": "The decision is recorded, but publication is incomplete. Check the current state and retry publication.",
-  "baseline_review_action_approved": "The decision is recorded, but publication is incomplete. Check the current state and retry publication.",
-  "baseline_review_action_consumed": "The baseline is published in the catalog.",
+  "baseline_review_action_confirmed": "The decision is recorded, but the catalog entry was not saved. Check the current state and retry saving.",
+  "baseline_review_action_approved": "The decision is recorded, but the catalog entry was not saved. Check the current state and retry saving.",
+  "baseline_review_action_consumed": "The baseline is approved in the local catalog. Git publication is a separate step.",
  "baseline_review_action_denied": "The candidate was rejected. The reason is stored with the decision.",
  "baseline_review_action_unknown": "Reload the list before taking another action.",
  "baseline_review_decision_failed": "The decision did not finish.",
@@ -575,9 +575,9 @@
  "baseline_review_unbound": "No authoritative release",
  "baseline_review_unbound_help": "This candidate has no linked release; capture it against a release to enable approval.",
  "baseline_review_reason": "Review reason",
-  "baseline_review_approve": "Approve and publish",
+  "baseline_review_approve": "Approve and save baseline",
  "baseline_review_reject": "Reject",
-  "baseline_review_publish": "Retry publication",
+  "baseline_review_publish": "Retry saving baseline",
  "reference_title": "Explore a Superset reference dashboard",
  "reference_intro": "Paste a URL for this scenario's dashboard to inspect its filters and metric values.",
  "reference_url_label": "Reference dashboard URL",
--- a/frontend/src/lib/i18n/locales/ru/dashboard-testing.json
+++ b/frontend/src/lib/i18n/locales/ru/dashboard-testing.json
@@ -523,12 +523,12 @@
  "baseline_review_loading": "Загрузка кандидатов…",
  "baseline_review_error": "Не удалось загрузить кандидатов baseline.",
  "baseline_review_recovery": "Обновите список, чтобы увидеть актуальное состояние.",
-  "baseline_review_missing": "Новых кандидатов для проверки нет. Состояние опубликованных значений показано выше; для нового решения нужен захват кандидата из запуска сценария.",
+  "baseline_review_missing": "Новых кандидатов для проверки нет. Утверждённые значения каталога показаны выше; для нового решения нужен захват кандидата из запуска сценария.",
  "baseline_overview_title": "Утверждённые значения",
  "baseline_overview_no_release": "Для этого дашборда нет утверждённого релиза. Сначала утвердите релиз, затем захватите кандидат baseline.",
-  "baseline_overview_missing": "Утверждённый релиз есть, но baseline для него ещё не опубликован.",
-  "baseline_overview_ready": "Опубликованные значения доступны для этого релиза.",
-  "baseline_overview_stale": "Некоторые опубликованные значения относятся к другому релизу и требуют проверки.",
+  "baseline_overview_missing": "Утверждённый релиз есть, но baseline для него ещё не утверждён.",
+  "baseline_overview_ready": "Утверждённые значения каталога доступны для этого релиза.",
+  "baseline_overview_stale": "Некоторые утверждённые значения каталога относятся к другому релизу и требуют проверки.",
  "baseline_overview_entry_stale": "Значение относится к другому релизу",
  "baseline_overview_more": "Ещё записей",
  "baseline_review_status": "Статус",
@@ -536,14 +536,14 @@
  "baseline_review_state_pending": "Ожидает решения",
  "baseline_review_state_confirmed": "Решение записано",
  "baseline_review_state_approved": "Решение записано",
-  "baseline_review_state_consumed": "Опубликован",
+  "baseline_review_state_consumed": "Утверждён в каталоге",
  "baseline_review_state_denied": "Отклонён",
  "baseline_review_state_unknown": "Состояние требует проверки",
  "baseline_review_action_draft": "Сверьте источник, значение и правило сравнения. Затем укажите причину решения.",
  "baseline_review_action_pending": "Сверьте источник, значение и правило сравнения. Затем укажите причину решения.",
-  "baseline_review_action_confirmed": "Решение записано. Публикация ещё не завершена; повторите её после проверки текущего состояния.",
-  "baseline_review_action_approved": "Решение записано. Публикация ещё не завершена; повторите её после проверки текущего состояния.",
-  "baseline_review_action_consumed": "Baseline опубликован в каталоге.",
+  "baseline_review_action_confirmed": "Решение записано, но запись каталога не сохранена. Проверьте текущее состояние и повторите сохранение.",
+  "baseline_review_action_approved": "Решение записано, но запись каталога не сохранена. Проверьте текущее состояние и повторите сохранение.",
+  "baseline_review_action_consumed": "Baseline утверждён в локальном каталоге. Публикация в Git выполняется отдельно.",
  "baseline_review_action_denied": "Кандидат отклонён. Причина сохранена в истории решения.",
  "baseline_review_action_unknown": "Обновите список перед следующим действием.",
  "baseline_review_decision_failed": "Решение не завершено.",
@@ -575,9 +575,9 @@
  "baseline_review_unbound": "Нет авторитетного релиза",
  "baseline_review_unbound_help": "Кандидат не связан с релизом; для утверждения захватите его с привязкой к релизу.",
  "baseline_review_reason": "Причина решения",
-  "baseline_review_approve": "Утвердить и опубликовать",
+  "baseline_review_approve": "Утвердить и сохранить baseline",
  "baseline_review_reject": "Отклонить",
-  "baseline_review_publish": "Повторить публикацию",
+  "baseline_review_publish": "Повторить сохранение baseline",
  "reference_title": "Просмотр эталонного дашборда Superset",
  "reference_intro": "Вставьте URL дашборда этого сценария, чтобы проверить фильтры и значения метрик.",
  "reference_url_label": "URL эталонного дашборда",
--- a/frontend/src/lib/types/scenario-automation.ts
+++ b/frontend/src/lib/types/scenario-automation.ts
@@ -17,6 +17,8 @@ export interface ScenarioSchedule {
  scenario_id: string;
  revision_policy: RevisionPolicy;
  revision_id: string | null;
+  baseline_set: string | null;
+  baseline_set_version: string | null;
  environment_id: string;
  cron_expr: string;
  timezone: string;
@@ -142,6 +144,8 @@ export interface ScheduleUpsertInput {
  timezone: string;
  revision_policy: RevisionPolicy;
  revision_id: string | null;
+  baseline_set: string | null;
+  baseline_set_version: string | null;
  policy_id: string | null;
  missed_execution_policy: MissedExecutionPolicy;
  enabled: boolean;