test(mcp): REST/MCP error-shape parity matrix with a shared start-error classifier (050 T051)

- start_run.py: classify_start_error — one canonical mapper for ENVIRONMENT_NOT_CONFIGURED,
  every BASELINE_* code, IDEMPOTENCY_KEY_REUSED and existing UPPER_SNAKE typed codes; unknown
  ValueErrors collapse to RUN_START_CONFLICT. REST and MCP now share it, so the same failure
  can no longer present a transport-dependent error vocabulary (the divergence the matrix caught:
  MCP returned raw exception text while REST collapsed to RUN_START_CONFLICT).
- Parity matrix (12+ vectors): env/baseline/conflict/disabled-action/disabled-automation/
  investigation-RBAC/unauthenticated, plus service-principal human-gate denial.
- Base-drift adaptations during integration: the investigation-case ACL vector now asserts the
  real no-bypass invariant (foreign case collapses to not_found via MCP, no content leak) since
  the investigation MCP tools exist on this release; terminal-transition atomicity expectations
  now include the queue_created receipt and pin replay idempotency (2 receipts, no growth).
- 17 parity tests; union wave-3 verification: 1845 passed.

backend/src/api/routes/dashboard_testing/scenario_automation.py

@@ -53,7 +53,7 @@ from src.services.dashboard_testing.automation.trigger import dispatch_trigger_e
 from src.services.dashboard_testing.execution.environment_policy import (
     resolve_environment_execution_policy,
 )
-from src.services.dashboard_testing.execution.runner import start_run
+from src.services.dashboard_testing.execution.runner import classify_start_error, start_run
 
 router = APIRouter(prefix="/api/scenario-automation", tags=["scenario-automation"])
 _DB = Depends(get_db)
@@ -604,11 +604,7 @@ def api_trigger_scenario(
         raise HTTPException(status_code=403, detail={"code": "PROD_APPROVAL_REQUIRED", "detail": str(exc)}) from exc
     except ValueError as exc:
         db.rollback()
-        code = (
-            "AUTOMATION_INELIGIBLE_HUMAN_STEP"
-            if str(exc) == "AUTOMATION_INELIGIBLE_HUMAN_STEP"
-            else "IDEMPOTENCY_KEY_REUSED" if "IDEMPOTENCY" in str(exc) else "RUN_START_CONFLICT"
-        )
+        code = classify_start_error(exc)
         raise HTTPException(status_code=409, detail={"code": code, "detail": str(exc)}) from exc
     return {"run_id": run.id, "status": run.status, "scenario_id": scenario_id}
 # #endregion Api.ScenarioAutomation.DirectTrigger

backend/src/api/routes/dashboard_testing/scenario_runs.py

@@ -39,6 +39,7 @@ from src.services.dashboard_testing.execution.lifecycle import (
 from src.services.dashboard_testing.execution.result import build_result
 from src.services.dashboard_testing.execution.baseline_resolver import BASELINE_RESOLVE_ERRORS
 from src.services.dashboard_testing.execution.runner import (
+    classify_start_error,
     continue_after_human_decision,
     continue_after_infrastructure_resume,
     continue_after_step_retry,
@@ -118,7 +119,13 @@ def _get_steps(db, run_id: str) -> list[ScenarioStepRun]:
 
 
 def _require_run_prod(current_user) -> None:
-    has_permission("scenario", "RUN_PROD")(current_user)
+    # has_permission still owns the RBAC decision and security log; converting its HTTPException
+    # into PermissionError lets the start handler emit the documented typed PROD_APPROVAL_REQUIRED
+    # envelope instead of a bare permission string (050 T044 error-shape parity).
+    try:
+        has_permission("scenario", "RUN_PROD")(current_user)
+    except HTTPException as exc:
+        raise PermissionError("scenario:run_prod required") from exc
 
 
 # #region Api.ScenarioExecution.Routes.Start [C:4] [TYPE Function] [SEMANTICS api,scenario,run,start,idempotency,rbac]
@@ -126,7 +133,8 @@ def _require_run_prod(current_user) -> None:
 # @BRIEF POST /api/scenario-runs — create a run (Idempotency-Key); PROD requires scenario:run:prod.
 # @RELATION CALLS -> [ScenarioExecution.Runner.Start]
 # @RELATION CALLS -> [ScenarioExecution.EnvironmentPolicy.Resolve]
-# @POST 201 with queued/pending-approval run; 409 IDEMPOTENCY_KEY_REUSED / RUN_START_CONFLICT;
+# @POST 201 with queued/pending-approval run; 409 with the canonical classify_start_error code
+#       (IDEMPOTENCY_KEY_REUSED / RUN_START_CONFLICT / an existing UPPER_SNAKE start code);
 #       403 on missing RBAC; 422 ENVIRONMENT_NOT_CONFIGURED / BASELINE_* before any durable side effect.
 # @INVARIANT This authenticated analyst route supplies the trusted manual origin itself; clients
 #            cannot select trigger_source in StartRunRequest.
@@ -169,12 +177,12 @@ def api_start_run(
         raise HTTPException(status_code=403, detail={"code": "PROD_APPROVAL_REQUIRED", "detail": str(exc)}) from exc
     except ValueError as exc:
         db.rollback()
-        if str(exc) == "ENVIRONMENT_NOT_CONFIGURED" or str(exc) in BASELINE_RESOLVE_ERRORS:
+        code = classify_start_error(exc)
+        if code == "ENVIRONMENT_NOT_CONFIGURED" or code in BASELINE_RESOLVE_ERRORS:
             raise HTTPException(
                 status_code=422,
-                detail={"code": str(exc), "detail": str(exc)},
+                detail={"code": code, "detail": str(exc)},
             ) from exc
-        code = "IDEMPOTENCY_KEY_REUSED" if "IDEMPOTENCY" in str(exc) else "RUN_START_CONFLICT"
         raise HTTPException(status_code=409, detail={"code": code, "detail": str(exc)}) from exc
 # #endregion Api.ScenarioExecution.Routes.Start
 

backend/src/mcp_server/rbac_server.py

@@ -57,16 +57,32 @@ def _gate_arguments(arguments: dict[str, Any]) -> dict[str, Any]:
 #            mcp_invocation gate or route an approved continuation into a new pending run.
 # @REJECTED An MCP-level approval gate for PROD scenario starts was rejected because it duplicates
 #           start_run's authoritative gate and can complete the invocation while the run is pending.
+class ScenarioStartPolicyUnavailable(ValueError):
+    """Start-gate classification failure carrying the typed code REST would return.
+
+    @POST `.code` is ENVIRONMENT_NOT_CONFIGURED for an unknown target and
+          scenario_start_policy_unavailable when the request body itself is invalid.
+    """
+
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = code
+
+
 def _scenario_start_permission(arguments: dict[str, Any]) -> tuple[str, str]:
     try:
         request = ScenarioStartInput.model_validate(_gate_arguments(arguments))
+    except ValidationError:
+        raise ScenarioStartPolicyUnavailable("scenario_start_policy_unavailable")
+    try:
         is_prod = resolve_environment_execution_policy(
             request.environment_id, get_config_manager()
         ).is_prod
-        return ("scenario", "RUN_PROD" if is_prod else "RUN")
-    except (ValidationError, ValueError):
-        # Invalid or unknown environments must not be converted into a weaker permission.
-        raise ValueError("scenario_start_policy_unavailable")
+    except ValueError as exc:
+        # 050 T044 residual: preserve the policy code (ENVIRONMENT_NOT_CONFIGURED) so the MCP
+        # envelope can match the REST 422 instead of collapsing to an opaque permission_denied.
+        raise ScenarioStartPolicyUnavailable(str(exc)) from exc
+    return ("scenario", "RUN_PROD" if is_prod else "RUN")
 # #endregion McpServer.ScenarioGate
 
 
@@ -314,15 +330,14 @@ class RbacFastMCP(FastMCP):
     async def call_tool(self, name: str, arguments: dict[str, Any]):
         definition = _MCP_CATALOG_BY_NAME.get(name)
         required_permission = definition.permission if definition is not None else None
+        policy_error: str | None = None
         if name == "start_scenario_run":
             try:
                 required_permission = _scenario_start_permission(arguments)
-            except ValueError:
+            except ScenarioStartPolicyUnavailable as exc:
+                policy_error = exc.code
                 required_permission = None
-        if not self._can_use_tool(name) or (
-            name == "start_scenario_run"
-            and (required_permission is None or not self._has_permission(required_permission))
-        ):
+        if not self._can_use_tool(name):
             self._record(
                 operation="tools/call",
                 tool_name=name,
@@ -331,6 +346,37 @@ class RbacFastMCP(FastMCP):
                 error_code="permission_denied",
             )
             raise PermissionError("permission_denied")
+        if name == "start_scenario_run":
+            if required_permission is None:
+                # 050 T044 residual: an unclassifiable target (unknown/unconfigured environment)
+                # keeps the fail-closed gate, but surfaces the same typed code as the REST 422
+                # instead of an opaque permission_denied. Invalid bodies stay permission-denied.
+                if policy_error == "ENVIRONMENT_NOT_CONFIGURED":
+                    self._record(
+                        operation="tools/call",
+                        tool_name=name,
+                        arguments=arguments,
+                        outcome="denied",
+                        error_code="environment_not_configured",
+                    )
+                    return {"status": "blocked", "error": "ENVIRONMENT_NOT_CONFIGURED"}
+                self._record(
+                    operation="tools/call",
+                    tool_name=name,
+                    arguments=arguments,
+                    outcome="denied",
+                    error_code="permission_denied",
+                )
+                raise PermissionError("permission_denied")
+            if not self._has_permission(required_permission):
+                self._record(
+                    operation="tools/call",
+                    tool_name=name,
+                    arguments=arguments,
+                    outcome="denied",
+                    error_code="permission_denied",
+                )
+                raise PermissionError("permission_denied")
         definition = _MCP_CATALOG_BY_NAME[name]
         # start_run creates the sole approval-gated PROD path. MCP approval dispatch is
         # reserved for tools whose catalog definition explicitly requires an MCP gate.

backend/src/mcp_server/tools_automation.py

@@ -205,7 +205,7 @@ def register_automation_tools(server: Any) -> None:
         owner()
         with SessionLocal() as db:
             if request.missed_execution_policy not in _MISSED_POLICIES:
-                raise ValueError("invalid missed_execution_policy")
+                raise ValueError("INVALID_MISSED_POLICY")
             _revision(db, request)
             _policy(db, request.policy_id)
             replay = _replay(db, ScenarioSchedule, request)
@@ -276,7 +276,7 @@ def register_automation_tools(server: Any) -> None:
         owner()
         with SessionLocal() as db:
             if request.trigger not in _TRIGGER_TYPES:
-                raise ValueError("invalid trigger")
+                raise ValueError("INVALID_TRIGGER")
             _revision(db, request)
             _policy(db, request.policy_id)
             replay = _replay(db, ScenarioTriggerRule, request)

backend/src/mcp_server/tools_scenario.py

@@ -58,7 +58,7 @@ from src.services.dashboard_testing.scenario.handles import (
 )
 from src.services.dashboard_testing.scenario.resolver import ResolveChange, resolve_scenario
 from src.services.dashboard_testing.scenario.validator import validate_scenario
-from src.services.dashboard_testing.execution.runner import start_run
+from src.services.dashboard_testing.execution.runner import classify_start_error, start_run
 from src.services.health_service import HealthService
 from src.services.llm_provider import LLMProviderService
 from src.services.mcp_approvals import decide_mcp_approval, list_pending_mcp_approvals
@@ -410,7 +410,9 @@ def register_scenario_tools(server) -> None:
             except (ValueError, PermissionError) as exc:
                 db.rollback()
                 logger.explore("Scenario start rejected", src="McpServer.ScenarioTools.start_scenario_run", error=str(exc))
-                return {"status": "blocked", "error": str(exc)}
+                # 050 T044 residual: the same classify_start_error used by REST keeps the typed code
+                # identical across transports; detail preserves the raw cause for operators.
+                return {"status": "blocked", "error": classify_start_error(exc), "detail": str(exc)}
 
     # #endregion McpServer.ScenarioTools
 

backend/src/services/dashboard_testing/execution/runner.py

@@ -41,6 +41,7 @@ from .start_run import (  # noqa: F401
     TRIGGER_SOURCE_MANUAL,
     TRIGGER_SOURCE_RELEASE,
     TRIGGER_SOURCE_SCHEDULED,
+    classify_start_error,
     start_run,
 )
 from .capacity_block import (  # noqa: F401

backend/src/services/dashboard_testing/execution/start_run.py

@@ -24,6 +24,7 @@ from __future__ import annotations
 import hashlib
 from datetime import UTC
 import json
+import re
 from typing import Any
 import uuid
 
@@ -34,7 +35,12 @@ from src.models.scenario_registry import ScenarioRegistryEntry, ScenarioRevision
 from src.models.scenario_run import ScenarioRun
 
 from .approval import create_prod_gate
-from .baseline_resolver import attach_baseline_pin, load_published_catalog, resolve_baseline_pin
+from .baseline_resolver import (
+    BASELINE_RESOLVE_ERRORS,
+    attach_baseline_pin,
+    load_published_catalog,
+    resolve_baseline_pin,
+)
 from .decision_policy import policy_digest
 from .environment_policy import resolve_environment_execution_policy
 from .live_binding import LiveExecutionBinding
@@ -376,4 +382,30 @@ def start_run(
     return run
 # #endregion ScenarioExecution.Runner.Start
 
+
+# #region ScenarioExecution.StartError.Classify [C:3] [TYPE Function] [SEMANTICS scenario,execution,start,error,parity]
+# @ingroup ScenarioExecution
+# @BRIEF Map one start-boundary failure to the canonical typed code every surface must return.
+# @PRE exc is the ValueError/PermissionError raised by start_run or its policy/plan/baseline resolvers.
+# @POST Returns ENVIRONMENT_NOT_CONFIGURED and every BASELINE_* code verbatim, IDEMPOTENCY_KEY_REUSED
+#       for any idempotency failure, an existing UPPER_SNAKE typed code verbatim, else RUN_START_CONFLICT.
+# @INVARIANT REST and MCP share this single classifier, so the same failure can never present a
+#            transport-dependent error vocabulary (050 T044 residual).
+# @RATIONALE The MCP tool returned the raw exception text while the REST route collapsed unknown
+#            ValueErrors to RUN_START_CONFLICT; one classifier removes that divergence by construction.
+# @REJECTED Keeping a private mapping in each transport was rejected — the two mappings already drifted.
+_START_ERROR_CODE = re.compile(r"^[A-Z][A-Z0-9_]*$")
+
+
+def classify_start_error(exc: BaseException) -> str:
+    message = str(exc)
+    if message == "ENVIRONMENT_NOT_CONFIGURED" or message in BASELINE_RESOLVE_ERRORS:
+        return message
+    if "IDEMPOTENCY" in message:
+        return "IDEMPOTENCY_KEY_REUSED"
+    if _START_ERROR_CODE.match(message):
+        return message
+    return "RUN_START_CONFLICT"
+# #endregion ScenarioExecution.StartError.Classify
+
 # #endregion ScenarioExecution.StartRun

backend/tests/services/dashboard_testing/registry/test_due_admission_atomicity.py

@@ -290,11 +290,14 @@ def test_terminal_transition_queue_and_outbox_share_one_transaction(registry_ses
     assert terminal.phase == "completed"
     assert registry_session.query(InvestigationQueueItem).filter_by(run_id=_TERMINAL_RUN_ID).count() == 1
     notifications = registry_session.query(ScenarioNotificationEvent).filter_by(run_id=_TERMINAL_RUN_ID).all()
-    assert [event.event_type for event in notifications] == ["failed"]
+    # Terminal failure emits the terminal `failed` receipt; projecting the new queue item emits its
+    # own idempotent `queue_created` receipt (NOTIFY-001 wiring) in the same transaction.
+    assert sorted(event.event_type for event in notifications) == ["failed", "queue_created"]
 
     scheduler_module.execute_scheduled_queued_scenario_dispatch()
     assert registry_session.query(InvestigationQueueItem).filter_by(run_id=_TERMINAL_RUN_ID).count() == 1
-    assert registry_session.query(ScenarioNotificationEvent).filter_by(run_id=_TERMINAL_RUN_ID).count() == 1
+    # Replay is idempotent: neither the queue projection nor either receipt duplicates.
+    assert registry_session.query(ScenarioNotificationEvent).filter_by(run_id=_TERMINAL_RUN_ID).count() == 2
 # #endregion Test.ScenarioExecution.DueAdmissionAtomicity.TerminalOutbox
 
 

backend/tests/test_mcp_rest_error_parity.py

@@ -0,0 +1,693 @@
+# #region Test.McpRestErrorParity [C:5] [TYPE Module] [SEMANTICS test,mcp,parity,rest,error,lifecycle,baseline,rbac,disabled,investigation]
+# @BRIEF 050 T044 residual: one hardcoded offline matrix proves the REST route and the MCP tool
+#        return the same typed error code for lifecycle, baseline, auth, disabled-automation and
+#        investigation-boundary cases.
+# @RELATION VERIFIES -> [McpServer.ScenarioTools]
+# @RELATION VERIFIES -> [McpServer.RbacServer]
+# @RELATION VERIFIES -> [Api.ScenarioExecution.Routes.Start]
+# @RELATION VERIFIES -> [Api.ScenarioAutomation.Routes]
+# @RELATION VERIFIES -> [Api.ScenarioAnalytics.Routes]
+# @RELATION VERIFIES -> [ScenarioExecution.StartError.Classify]
+# @TEST_INVARIANT transport_parity -> each case's REST typed code equals the MCP typed code (or its
+#                 documented permission equivalence), because both share classify_start_error.
+#                 -> VERIFIED_BY: every test_*_parity case below.
+# @TEST_INVARIANT production_gate_parity -> a PROD start is gated on both surfaces; the same
+#                 principal reaches pending_approval on both. -> VERIFIED_BY: test_prod_start_*.
+# @TEST_EDGE no_investigation_bypass -> the MCP catalogue exposes no case/queue tool, so the REST
+#           object ACL cannot be bypassed over MCP.
+# @TEST_FIXTURE hardcoded graphs + publishedCatalog snapshots -> built in this module, no live stand.
+# @RATIONALE The MCP tool once returned raw exception text while REST collapsed conflicts to
+#            RUN_START_CONFLICT; a shared classifier plus this matrix pin the two transports together.
+from __future__ import annotations
+
+import asyncio
+import secrets
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+from fastapi.testclient import TestClient
+
+from src.app import app
+from src.core.auth.security import get_password_hash
+from src.core.database import SessionLocal
+from src.dependencies import get_config_manager, get_current_user
+from src.mcp_server import server as mcp_server
+from src.mcp_server.server import (
+    _MCP_CATALOG_BY_NAME,
+    _access_token_context,
+    create_mcp_asgi_app,
+)
+import src.mcp_server.rbac_server as rbac_server_module
+import src.mcp_server.tools_scenario as tools_scenario_module
+from src.models.auth import Permission, Role, User
+import src.models.scenario_investigation as investigation_models
+from src.models.scenario_registry import ScenarioRegistryEntry, ScenarioRevision
+from src.models.scenario_run import ScenarioRun
+from src.services.dashboard_testing.scenario.templates import (
+    ACTION_REGISTRY_VERSION,
+    action_registry_fingerprint,
+    resolve_action_descriptor,
+)
+
+_ENVIRONMENTS = {
+    "env-dev": SimpleNamespace(id="env-dev", stage="DEV", is_production=False),
+    "env-prod": SimpleNamespace(id="env-prod", stage="PROD", is_production=True),
+}
+_INVESTIGATION_TOOL_TOKENS = ("investigation", "case", "queue")
+_HEX40 = "c" * 40
+_HEX64 = "a" * 64
+_HEX64B = "b" * 64
+_BASELINE_SET = "ss-parity-set"
+_BASELINE_VERSION = "7"
+
+
+# #region Test.McpRestErrorParity.Fixtures [C:4] [TYPE Function]
+# @ingroup Test.McpRestErrorParity
+# @BRIEF Hardcoded graphs, catalogue snapshots and the shared principal/DB harness.
+def _config_manager() -> SimpleNamespace:
+    return SimpleNamespace(get_environment=lambda env_id: _ENVIRONMENTS.get(env_id))
+
+
+def _patch_config(monkeypatch) -> None:
+    manager = _config_manager()
+    monkeypatch.setattr(rbac_server_module, "get_config_manager", lambda: manager)
+    monkeypatch.setattr(tools_scenario_module, "get_config_manager", lambda: manager)
+
+
+def _step(step_id: str, tool: str, action: str, **extra) -> dict:
+    return {
+        "logical_step_id": step_id,
+        "tool": tool,
+        "action": action,
+        "action_descriptor": resolve_action_descriptor(
+            tool=tool,
+            action=action,
+            registry_version=ACTION_REGISTRY_VERSION,
+            registry_hash=action_registry_fingerprint(),
+        ).snapshot(),
+        **extra,
+    }
+
+
+def _plain_graph() -> dict:
+    return {
+        "schema_version": 1,
+        "action_registry_version": ACTION_REGISTRY_VERSION,
+        "action_registry_hash": action_registry_fingerprint(),
+        "environment_ids": ["env-dev"],
+        "steps": [_step("s1_assert", "assertion", "structural_assert", actual=1, expected=1)],
+        "dependencies": [],
+    }
+
+
+def _baseline_graph() -> dict:
+    graph = _plain_graph()
+    graph["baselines"] = {"revenue_sum": {"reference": "baseline-v1", "policy": "exact"}}
+    return graph
+
+
+def _human_graph() -> dict:
+    return {
+        "schema_version": 1,
+        "action_registry_version": ACTION_REGISTRY_VERSION,
+        "action_registry_hash": action_registry_fingerprint(),
+        "environment_ids": ["env-dev"],
+        "steps": [_step("s1_human", "human", "human_checkpoint")],
+        "dependencies": [],
+    }
+
+
+def _approved_entry(baseline_id: str = "bl-1", **overrides) -> dict:
+    item = {
+        "entry": {
+            "kind": "metric",
+            "release_version": "v1.0.0",
+            "release_commit_hash": _HEX40,
+            "source_response_hash": _HEX64,
+        },
+        "baseline_id": baseline_id,
+        "baseline_revision_id": f"{baseline_id}-rev",
+        "coordinate_hash": _HEX64,
+        "entry_digest": _HEX64,
+        "capture_artifact_id": "artifact-1",
+        "capture_profile_hash": _HEX64,
+        "status": "approved",
+    }
+    item.update(overrides)
+    return item
+
+
+def _catalog_snapshot(
+    *,
+    revision_digest: str = _HEX64,
+    entries: list[dict] | None = None,
+) -> dict:
+    return {
+        "baseline_set_id": _BASELINE_SET,
+        "baseline_set_version": _BASELINE_VERSION,
+        "catalog_digest": _HEX64,
+        "release_id": "rel-1",
+        "baseline_family": _HEX64B,
+        "catalog_revision": {
+            "catalog_revision_id": "cr-1",
+            "catalog_digest": revision_digest,
+            "publication": {
+                "state": "published",
+                "commit_hash": _HEX40,
+                "published_receipt_id": "rcpt-1",
+            },
+            "entry_revisions": entries if entries is not None else [_approved_entry()],
+        },
+    }
+
+
+def _start_body(scenario_id: str, revision_id: str, *, environment_id: str, **extra) -> dict:
+    return {
+        "scenario_id": scenario_id,
+        "revision_id": revision_id,
+        "environment_id": environment_id,
+        "params": {},
+        **extra,
+    }
+
+
+def _mcp_call(server, tool_name: str, arguments: dict) -> dict:
+    async def _run():
+        return await server.call_tool(tool_name, arguments)
+
+    loop = asyncio.new_event_loop()
+    try:
+        try:
+            result = loop.run_until_complete(_run())
+        except Exception as exc:  # transport/tool error is the assertion target, not a bug
+            return {"status": "raised", "error": str(exc)}
+    finally:
+        loop.close()
+    return result[1] if isinstance(result, tuple) else result
+
+
+class ParityEnv:
+    """Shared global-DB harness: REST over src.app, MCP over the probe server."""
+
+    def __init__(self) -> None:
+        self.scenario_ids: list[str] = []
+        self.usernames: list[str] = []
+        self.role_names: list[str] = []
+        self.run_ids: list[str] = []
+
+    def seed_scenario(self, graph: dict) -> tuple[str, str]:
+        scenario_id = str(uuid4())
+        revision_id = str(uuid4())
+        with SessionLocal() as db:
+            db.add(ScenarioRegistryEntry(
+                scenario_id=scenario_id, scenario_key=f"parity-{scenario_id[:8]}",
+                name="parity scenario", dashboard_id=80, environment_ids=["env-dev"],
+                owner_id="owner", owner_username="owner",
+                lifecycle_status="READY", validation_status="valid",
+                current_revision_id=revision_id,
+            ))
+            self._add_revision(db, scenario_id, revision_id, graph, "current")
+            db.commit()
+        self.scenario_ids.append(scenario_id)
+        return scenario_id, revision_id
+
+    def add_stale_revision(self, scenario_id: str) -> str:
+        revision_id = str(uuid4())
+        with SessionLocal() as db:
+            self._add_revision(db, scenario_id, revision_id, _plain_graph(), "candidate")
+            db.commit()
+        return revision_id
+
+    @staticmethod
+    def _add_revision(db, scenario_id: str, revision_id: str, graph: dict, activation: str) -> None:
+        db.add(ScenarioRevision(
+            revision_id=revision_id, scenario_id=scenario_id, content_hash=_HEX64,
+            graph_snapshot=graph, execution_template_hash="", template_version="v1",
+            schema_version=1, compatibility_family="default", change_summary={},
+            created_by="owner", activation_status=activation,
+        ))
+
+    def seed_user(self, *perms: tuple[str, str], admin: bool = False) -> tuple[str, str]:
+        suffix = secrets.token_hex(4)
+        username = f"parity-user-{suffix}"
+        role_name = f"ParityRole-{suffix}"
+        role = Role(
+            name=role_name, is_admin=admin,
+            permissions=[Permission(resource=r, action=a) for r, a in perms],
+        )
+        user = User(username=username, password_hash=get_password_hash("pw"), is_active=True, roles=[role])
+        with SessionLocal() as db:
+            db.add(user)
+            db.commit()
+            user_id = str(user.id)
+        self.usernames.append(username)
+        self.role_names.append(role_name)
+        return username, user_id
+
+    def rest_client(self, username: str, user_id: str, *perms: tuple[str, str], admin: bool = False) -> TestClient:
+        principal = SimpleNamespace(
+            id=user_id, username=username,
+            roles=[SimpleNamespace(
+                is_admin=admin,
+                permissions=[SimpleNamespace(resource=r, action=a) for r, a in perms],
+            )],
+        )
+        app.dependency_overrides[get_current_user] = lambda: principal
+        app.dependency_overrides[get_config_manager] = _config_manager
+        return TestClient(app)
+
+    def mcp_as(self, username: str | None, principal_type: str = "user"):
+        server = mcp_server._build_probe_server()
+        token = _access_token_context.set(mcp_server.AccessToken(
+            token="parity-token", client_id="parity-client", scopes=["mcp"],
+            subject=username if principal_type == "user" else None,
+            claims={"principal_type": principal_type},
+        ))
+        return server, token
+
+    def track_runs(self, prefix: str) -> None:
+        with SessionLocal() as db:
+            rows = db.query(ScenarioRun).filter(ScenarioRun.idempotency_key.like(f"{prefix}%")).all()
+            self.run_ids.extend(row.id for row in rows)
+
+    def cleanup(self) -> None:
+        app.dependency_overrides.pop(get_current_user, None)
+        app.dependency_overrides.pop(get_config_manager, None)
+        with SessionLocal() as db:
+            if self.run_ids:
+                db.query(ScenarioRun).filter(ScenarioRun.id.in_(self.run_ids)).delete(synchronize_session=False)
+            for scenario_id in self.scenario_ids:
+                db.query(ScenarioRegistryEntry).filter_by(scenario_id=scenario_id).delete()
+                db.query(ScenarioRevision).filter_by(scenario_id=scenario_id).delete()
+            for username in self.usernames:
+                user = db.query(User).filter(User.username == username).first()
+                if user is not None:
+                    db.delete(user)
+                    db.flush()
+            for role_name in self.role_names:
+                role = db.query(Role).filter(Role.name == role_name).first()
+                if role is not None:
+                    db.delete(role)
+            db.commit()
+
+
+@pytest.fixture
+def parity():
+    env = ParityEnv()
+    try:
+        yield env
+    finally:
+        env.cleanup()
+# #endregion Test.McpRestErrorParity.Fixtures
+
+
+# #region Test.McpRestErrorParity.Lifecycle [C:4] [TYPE Function]
+# @ingroup Test.McpRestErrorParity
+# @BRIEF Lifecycle start errors carry one typed code on both transports.
+def test_environment_not_configured_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_plain_graph())
+    username, user_id = parity.seed_user(("scenario", "RUN"))
+    rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
+    body = _start_body(scenario_id, revision_id, environment_id="env-ghost")
+
+    rest_response = rest.post(
+        "/api/scenario-runs", headers={"Idempotency-Key": f"parity-env-{uuid4()}"}, json=body,
+    )
+    assert rest_response.status_code == 422
+    assert rest_response.json()["detail"]["code"] == "ENVIRONMENT_NOT_CONFIGURED"
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "start_scenario_run", {
+            "request": {**body, "idempotency_key": f"parity-env-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "blocked"
+    assert mcp_result["error"] == "ENVIRONMENT_NOT_CONFIGURED"
+
+
+def test_run_start_conflict_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, _ = parity.seed_scenario(_plain_graph())
+    stale_revision = parity.add_stale_revision(scenario_id)
+    username, user_id = parity.seed_user(("scenario", "RUN"))
+    rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
+    body = _start_body(scenario_id, stale_revision, environment_id="env-dev")
+
+    rest_response = rest.post(
+        "/api/scenario-runs", headers={"Idempotency-Key": f"parity-conflict-{uuid4()}"}, json=body,
+    )
+    assert rest_response.status_code == 409
+    assert rest_response.json()["detail"]["code"] == "RUN_START_CONFLICT"
+    assert "revision mismatch" in rest_response.json()["detail"]["detail"]
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "start_scenario_run", {
+            "request": {**body, "idempotency_key": f"parity-conflict-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "blocked"
+    assert mcp_result["error"] == "RUN_START_CONFLICT"
+    assert "revision mismatch" in mcp_result["detail"]
+
+
+def test_idempotency_key_reused_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_plain_graph())
+    username, user_id = parity.seed_user(("scenario", "RUN"))
+    rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
+    key = f"parity-idem-{uuid4()}"
+    parity.track_runs("parity-idem-")
+
+    first = rest.post(
+        "/api/scenario-runs", headers={"Idempotency-Key": key},
+        json=_start_body(scenario_id, revision_id, environment_id="env-dev", params={"n": 1}),
+    )
+    assert first.status_code == 201
+    second = rest.post(
+        "/api/scenario-runs", headers={"Idempotency-Key": key},
+        json=_start_body(scenario_id, revision_id, environment_id="env-dev", params={"n": 2}),
+    )
+    assert second.status_code == 409
+    assert second.json()["detail"]["code"] == "IDEMPOTENCY_KEY_REUSED"
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "start_scenario_run", {
+            "request": _start_body(scenario_id, revision_id, environment_id="env-dev", params={"n": 2})
+            | {"idempotency_key": key},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "blocked"
+    assert mcp_result["error"] == "IDEMPOTENCY_KEY_REUSED"
+
+
+def test_prod_approval_required_maps_to_mcp_permission(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_plain_graph())
+    username, user_id = parity.seed_user(("scenario", "RUN"))
+    rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
+    body = _start_body(scenario_id, revision_id, environment_id="env-prod")
+
+    rest_response = rest.post(
+        "/api/scenario-runs", headers={"Idempotency-Key": f"parity-prod-{uuid4()}"}, json=body,
+    )
+    assert rest_response.status_code == 403
+    assert rest_response.json()["detail"]["code"] == "PROD_APPROVAL_REQUIRED"
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "start_scenario_run", {
+            "request": {**body, "idempotency_key": f"parity-prod-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    # Documented equivalence: REST 403 PROD_APPROVAL_REQUIRED == MCP fail-closed permission denial.
+    assert mcp_result["status"] == "raised"
+    assert mcp_result["error"] == "permission_denied"
+
+
+def test_prod_start_reaches_pending_approval_on_both_surfaces(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_plain_graph())
+    username, user_id = parity.seed_user(("scenario", "RUN"), ("scenario", "RUN_PROD"))
+    rest = parity.rest_client(username, user_id, ("scenario", "RUN"), ("scenario", "RUN_PROD"))
+    body = _start_body(scenario_id, revision_id, environment_id="env-prod")
+
+    rest_response = rest.post(
+        "/api/scenario-runs", headers={"Idempotency-Key": f"parity-prod-ok-{uuid4()}"}, json=body,
+    )
+    assert rest_response.status_code == 201, rest_response.text
+    assert rest_response.json()["status"] == "pending_approval"
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "start_scenario_run", {
+            "request": {**body, "idempotency_key": f"parity-prod-ok-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "pending_approval"
+    parity.track_runs("parity-prod-ok-")
+# #endregion Test.McpRestErrorParity.Lifecycle
+
+
+# #region Test.McpRestErrorParity.Baseline [C:4] [TYPE Function]
+# @ingroup Test.McpRestErrorParity
+# @BRIEF Every 037 baseline-resolution code is identical on REST and MCP.
+@pytest.mark.parametrize(
+    "code,snapshot,with_selector",
+    [
+        ("BASELINE_MISSING", None, False),
+        ("BASELINE_NOT_PUBLISHED", {"catalog": "working-tree-yaml"}, True),
+        ("BASELINE_STALE", _catalog_snapshot(revision_digest="not-a-sha256"), True),
+        (
+            "BASELINE_AMBIGUOUS",
+            _catalog_snapshot(entries=[
+                _approved_entry("bl-dup", coordinate_hash="1" * 64),
+                _approved_entry("bl-dup", baseline_revision_id="bl-dup-rev-2", coordinate_hash="2" * 64),
+            ]),
+            True,
+        ),
+        (
+            "BASELINE_EVIDENCE_UNAVAILABLE",
+            _catalog_snapshot(entries=[
+                _approved_entry(**{"entry": {
+                    "kind": "metric", "release_version": "v1.0.0",
+                    "release_commit_hash": _HEX40, "source_response_hash": "short",
+                }}),
+            ]),
+            True,
+        ),
+    ],
+)
+def test_baseline_error_codes_parity(parity, monkeypatch, code, snapshot, with_selector) -> None:
+    monkeypatch.setattr(
+        "src.services.dashboard_testing.execution.start_run.load_published_catalog",
+        lambda injected=None: snapshot,
+    )
+    monkeypatch.setattr(
+        "src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
+        lambda *_args, **_kwargs: None,
+    )
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_baseline_graph())
+    username, user_id = parity.seed_user(("scenario", "RUN"))
+    rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
+    body = _start_body(scenario_id, revision_id, environment_id="env-dev")
+    if with_selector:
+        body["baseline_set"] = _BASELINE_SET
+        body["baseline_set_version"] = _BASELINE_VERSION
+
+    rest_response = rest.post(
+        "/api/scenario-runs", headers={"Idempotency-Key": f"parity-baseline-{uuid4()}"}, json=body,
+    )
+    assert rest_response.status_code == 422, rest_response.text
+    assert rest_response.json()["detail"]["code"] == code
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "start_scenario_run", {
+            "request": {"idempotency_key": f"parity-baseline-{uuid4()}", **{
+                k: v for k, v in body.items() if k != "idempotency_key"
+            }},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "blocked"
+    assert mcp_result["error"] == code
+# #endregion Test.McpRestErrorParity.Baseline
+
+
+# #region Test.McpRestErrorParity.DisabledAutomation [C:4] [TYPE Function]
+# @ingroup Test.McpRestErrorParity
+# @BRIEF Disabled/ineligible automation validation uses one code per failure on both surfaces.
+def test_automation_ineligible_human_step_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_human_graph())
+    username, user_id = parity.seed_user(admin=True)
+    rest = parity.rest_client(username, user_id, admin=True)
+
+    request_body = {
+        "scenario_id": scenario_id, "environment_id": "env-dev", "cron_expr": "0 * * * *",
+        "revision_policy": "pinned", "revision_id": revision_id,
+    }
+    rest_response = rest.post("/api/scenario-automation/schedules", json=request_body)
+    assert rest_response.status_code == 409
+    assert rest_response.json()["detail"]["code"] == "AUTOMATION_INELIGIBLE_HUMAN_STEP"
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "upsert_scenario_schedule", {
+            "request": {**request_body, "idempotency_key": f"parity-human-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "raised"
+    assert "AUTOMATION_INELIGIBLE_HUMAN_STEP" in mcp_result["error"]
+
+
+def test_invalid_missed_policy_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_plain_graph())
+    username, user_id = parity.seed_user(admin=True)
+    rest = parity.rest_client(username, user_id, admin=True)
+
+    request_body = {
+        "scenario_id": scenario_id, "environment_id": "env-dev", "cron_expr": "0 * * * *",
+        "revision_policy": "pinned", "revision_id": revision_id,
+        "missed_execution_policy": "bogus",
+    }
+    rest_response = rest.post("/api/scenario-automation/schedules", json=request_body)
+    assert rest_response.status_code == 422
+    assert rest_response.json()["detail"]["code"] == "INVALID_MISSED_POLICY"
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "upsert_scenario_schedule", {
+            "request": {**request_body, "idempotency_key": f"parity-missed-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "raised"
+    assert "INVALID_MISSED_POLICY" in mcp_result["error"]
+
+
+def test_invalid_trigger_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, revision_id = parity.seed_scenario(_plain_graph())
+    username, user_id = parity.seed_user(admin=True)
+    rest = parity.rest_client(username, user_id, admin=True)
+
+    request_body = {
+        "scenario_id": scenario_id, "environment_id": "env-dev", "trigger": "bogus",
+        "revision_policy": "pinned", "revision_id": revision_id,
+    }
+    rest_response = rest.post("/api/scenario-automation/trigger-rules", json=request_body)
+    assert rest_response.status_code == 422
+    assert rest_response.json()["detail"]["code"] == "INVALID_TRIGGER"
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "upsert_scenario_trigger_rule", {
+            "request": {**request_body, "idempotency_key": f"parity-trigger-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "raised"
+    assert "INVALID_TRIGGER" in mcp_result["error"]
+
+
+def test_non_active_revision_schedule_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, _ = parity.seed_scenario(_plain_graph())
+    stale_revision = parity.add_stale_revision(scenario_id)
+    username, user_id = parity.seed_user(admin=True)
+    rest = parity.rest_client(username, user_id, admin=True)
+
+    request_body = {
+        "scenario_id": scenario_id, "environment_id": "env-dev", "cron_expr": "0 * * * *",
+        "revision_policy": "pinned", "revision_id": stale_revision,
+    }
+    rest_response = rest.post("/api/scenario-automation/schedules", json=request_body)
+    assert rest_response.status_code == 409
+    expected = "revision must be the active scenario revision"
+    assert rest_response.json()["detail"]["code"] == expected
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "upsert_scenario_schedule", {
+            "request": {**request_body, "idempotency_key": f"parity-stale-{uuid4()}"},
+        })
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "raised"
+    assert expected in mcp_result["error"]
+# #endregion Test.McpRestErrorParity.DisabledAutomation
+
+
+# #region Test.McpRestErrorParity.AuthAndRbac [C:4] [TYPE Function]
+# @ingroup Test.McpRestErrorParity
+# @BRIEF Auth transport, human-gate decision authority and investigation ACL parity.
+def test_service_principal_cannot_decide_human_gate_parity(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    username, user_id = parity.seed_user(("scenario", "RUN"))
+    rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
+
+    # REST approval decision is gated on scenario RUN_PROD before the handler runs.
+    rest_response = rest.post(
+        f"/api/scenario-runs/{uuid4()}/approval/decision", json={"decision": "approve"},
+    )
+    assert rest_response.status_code == 403
+
+    # MCP: a service principal can never decide a human gate.
+    server, token = parity.mcp_as(None, principal_type="service")
+    try:
+        mcp_result = _mcp_call(server, "decide_approval", {"gate_id": "g-1", "decision": "approve"})
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["error"] == "permission_denied"
+
+
+def test_investigation_case_acl_has_no_mcp_bypass(parity, monkeypatch) -> None:
+    _patch_config(monkeypatch)
+    scenario_id, _ = parity.seed_scenario(_plain_graph())
+    username, user_id = parity.seed_user(("scenario:result", "VIEW"))
+    case_id = str(uuid4())
+    with SessionLocal() as db:
+        db.add(investigation_models.InvestigationCase(
+            id=case_id, fingerprint=f"fp-{case_id[:8]}", scenario_id=scenario_id,
+            status="open", decision_version=1, evidence_snapshot={}, linked_run_ids=[],
+            owner_id="other-owner",
+        ))
+        db.commit()
+
+    rest = parity.rest_client(username, user_id, ("scenario:result", "VIEW"))
+    rest_response = rest.get(f"/api/scenario-analytics/cases/{case_id}")
+    assert rest_response.status_code == 403
+    assert rest_response.json()["detail"]["code"] == "CASE_ACL_DENIED"
+
+    # Investigation MCP tools DO exist on this release (G-INVESTIGATION-MCP CLOSED); the parity
+    # requirement is that they cannot BYPASS the REST case ACL. A foreign case must collapse to
+    # not_found through MCP exactly as REST answers 403, with no case content disclosed.
+    offenders = [
+        name for name in _MCP_CATALOG_BY_NAME
+        if any(token in name.lower() for token in _INVESTIGATION_TOOL_TOKENS)
+    ]
+    assert "get_investigation_case" in offenders, offenders
+
+    server, token = parity.mcp_as(username)
+    try:
+        mcp_result = _mcp_call(server, "get_investigation_case", {"request": {"case_id": case_id}})
+    finally:
+        _access_token_context.reset(token)
+    assert mcp_result["status"] == "not_found", mcp_result
+    assert "fingerprint" not in mcp_result and "scenario_id" not in mcp_result, mcp_result
+
+    with SessionLocal() as db:
+        db.query(investigation_models.InvestigationCase).filter_by(id=case_id).delete()
+        db.commit()
+
+
+def test_unauthenticated_transport_parity() -> None:
+    rest_client = TestClient(app)
+    rest_response = rest_client.post(
+        "/api/scenario-runs",
+        headers={"Idempotency-Key": "parity-unauth"},
+        json={"scenario_id": "s", "revision_id": "r", "environment_id": "env-dev", "params": {}},
+    )
+    assert rest_response.status_code == 401
+
+    mcp_client = TestClient(create_mcp_asgi_app())
+    mcp_response = mcp_client.post("/", json={"jsonrpc": "2.0", "id": 1, "method": "ping"})
+    assert mcp_response.status_code == 401
+    assert mcp_response.json()["error"] == "authentication_required"
+# #endregion Test.McpRestErrorParity.AuthAndRbac
+
+# #endregion Test.McpRestErrorParity