Live 12 catalogue + Phase 2 output validator (warn-and-keep), symusic MIDI migration

apps/backend/phase2_catalogue_gates.py

@@ -0,0 +1,353 @@
+"""Live 12 source-catalogue checks that ANNOTATE Phase 2 recommendations.
+
+The Phase 2 Gemini handler emits recommendations as `{device, parameter, value,
+phase1Fields}` records inside `mixAndMasterChain`, `abletonRecommendations`,
+and `secretSauce.workflowSteps`. This module cross-checks each record against
+the source-extracted `data/live12_catalogue.json` (generated by
+`scripts/build_live12_catalogue.py` from the upstream MIDI Remote Scripts) and
+emits advisory `validationWarnings` for anything it cannot confirm.
+
+WARN-AND-KEEP CONTRACT (deliberate — see PR history):
+  This module NEVER drops a recommendation and NEVER rewrites a parameter. It
+  only emits warning-shaped events; `phase2_result` is left untouched.
+
+  Why warn instead of gate? Phase 2 is an *advisory* layer, the catalogue is
+  statically extracted and therefore incomplete, and Live's internal parameter
+  names ("1 Frequency A") diverge from a producer's natural phrasing ("Band 1
+  Frequency"). Dropping or auto-rewriting on that basis produced confidently-
+  wrong output: a fuzzy rewrite of "Band 1 Frequency" landed on "1 Frequency B"
+  — the wrong band AND the wrong A/B curve — and shipped it to the user with
+  full authority. That is worse than the gap it tried to close. So every check
+  now KEEPS the recommendation and rides a WARNING on the existing
+  `validationWarnings` channel. This mirrors the sibling
+  `server_phase2._validate_phase2_citation_paths`, which is likewise
+  WARNING-only and never mutates the Phase 1-authoritative payload.
+
+The checks (each warn-and-keep):
+
+  1. `device_unknown`     -- `device` is not in the source catalogue.
+  2. `parameter_unknown`  -- `parameter` is not on the device AND
+     `Live12Catalogue.fuzzy_resolve` finds NO close match. When fuzzy DOES find
+     a close match the parameter is almost certainly real but phrased
+     differently, so the record is kept SILENTLY (no warning). Fuzzy is used
+     here only to suppress false "unknown" warnings — NEVER to rewrite the
+     parameter, because the closest lexical match is frequently the wrong
+     instance on multi-band devices (the EQ-band failure above).
+  3. `value_out_of_range` -- numeric `value` falls outside the catalogue range.
+     Inert today: static extraction carries no `min`/`max`, so `has_range()` is
+     always False in production; reserved for future runtime-introspection
+     enrichment. Exercised only by tests that inject a ranged catalogue.
+  4. `citation_missing`   -- `phase1Fields` is missing or empty. The chain-of-
+     custody invariant requires every recommendation to cite >=1 Phase 1 field.
+
+Every warning carries `code="RECOMMENDATION_UNVERIFIED"`, a `reason`, plus
+`device`, `parameter`, `path`, and `requestId` so the operator-facing
+diagnostic log keeps a complete trail. The function returns the event list; the
+caller stitches it into the existing `validationWarnings` channel.
+
+Living separately from `server_phase2.py` keeps this free of the FastAPI /
+pydantic import chain so unit tests can exercise it in pure stdlib.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from math import isfinite
+from typing import Any
+
+from live12_catalogue import Live12Catalogue
+
+
+_REASON_DEVICE_UNKNOWN = "device_unknown"
+_REASON_PARAMETER_UNKNOWN = "parameter_unknown"
+_REASON_VALUE_OUT_OF_RANGE = "value_out_of_range"
+_REASON_CITATION_MISSING = "citation_missing"
+
+# Single warning code for every catalogue check. Nothing is rejected or
+# rewritten, so the old `RECOMMENDATION_REJECTED` / `PARAMETER_REWRITTEN` codes
+# are gone; consumers discriminate on the `reason` field. This code is
+# intentionally NOT in the frontend `SALVAGE_WARNING_CODES` set — warn-and-keep
+# means no recommendation was silently dropped or repaired, so it must not trip
+# the "Gemini output was salvaged" decision gate.
+_RECOMMENDATION_UNVERIFIED = "RECOMMENDATION_UNVERIFIED"
+
+_VALUE_NUMBER_RE = re.compile(r"-?\d+(?:\.\d+)?")
+
+
+def _as_record(value: Any) -> dict[str, Any] | None:
+    if not value or not isinstance(value, dict):
+        return None
+    return value
+
+
+def _stringify_warning_value(value: Any) -> str:
+    if isinstance(value, str):
+        return value
+    try:
+        return json.dumps(value, ensure_ascii=False, sort_keys=True)
+    except TypeError:
+        return str(value)
+
+
+def _coerce_value_to_number(raw: Any) -> float | None:
+    """Pull a finite numeric value out of Gemini's `value` field.
+
+    Gemini emits `value` as a free-form string ("4.5", "-12 dB", "10ms",
+    "1:4", "auto"). The range check only fires when we can extract a single
+    finite number; anything ambiguous (compound ratios, slashes, "auto", etc.)
+    passes through without a range check rather than producing a false-positive
+    warning.
+    """
+    if isinstance(raw, bool):  # bool is a subclass of int -- exclude.
+        return None
+    if isinstance(raw, (int, float)) and isfinite(float(raw)):
+        return float(raw)
+    if not isinstance(raw, str):
+        return None
+    if ":" in raw:
+        return None  # ratios like "1:4" cannot be one number
+    matches = _VALUE_NUMBER_RE.findall(raw)
+    if len(matches) != 1:
+        return None
+    try:
+        return float(matches[0])
+    except ValueError:
+        return None
+
+
+def _build_unverified_event(
+    *,
+    path: str,
+    reason: str,
+    message: str,
+    device: str,
+    parameter: str,
+    request_id: str,
+    value: Any = None,
+) -> dict[str, Any]:
+    event: dict[str, Any] = {
+        "code": _RECOMMENDATION_UNVERIFIED,
+        "path": path,
+        "message": message,
+        "reason": reason,
+        "device": device,
+        "parameter": parameter,
+        "requestId": request_id,
+    }
+    if value is not None:
+        event["value"] = _stringify_warning_value(value)
+    return event
+
+
+def _inspect_record(
+    *,
+    catalogue: Live12Catalogue,
+    record: dict[str, Any],
+    base_path: str,
+    request_id: str,
+    require_citation: bool,
+    events: list[dict[str, Any]],
+) -> None:
+    """Run advisory device/parameter/range/citation checks against one record.
+
+    Appends WARNING events to `events`. NEVER mutates `record` and NEVER signals
+    a drop — the recommendation stays in the user-facing payload regardless of
+    what the checks find.
+    """
+    device_raw = record.get("device")
+    parameter_raw = record.get("parameter")
+    device = device_raw.strip() if isinstance(device_raw, str) else ""
+    parameter = parameter_raw.strip() if isinstance(parameter_raw, str) else ""
+
+    canonical_device = catalogue.canonical_device(device)
+    if canonical_device is None:
+        events.append(
+            _build_unverified_event(
+                path=base_path,
+                reason=_REASON_DEVICE_UNKNOWN,
+                message=(
+                    f"Device {device!r} is not in the Live 12 source catalogue. "
+                    "Kept as-is, but flagged as unverified."
+                ),
+                device=device,
+                parameter=parameter,
+                request_id=request_id,
+            )
+        )
+    elif not catalogue.has_parameter(canonical_device, parameter):
+        # Not an exact catalogue parameter. Use fuzzy resolution ONLY to decide
+        # whether to warn: a close match means the parameter is almost certainly
+        # real under a slightly different name (e.g. "Band 1 Frequency" vs
+        # "1 Frequency A"), so keep it silently. The fuzzy *target* is NOT
+        # trusted enough to rewrite — on multi-band devices the closest lexical
+        # match is routinely the wrong instance.
+        if catalogue.fuzzy_resolve(canonical_device, parameter) is None:
+            events.append(
+                _build_unverified_event(
+                    path=base_path,
+                    reason=_REASON_PARAMETER_UNKNOWN,
+                    message=(
+                        f"Parameter {parameter!r} is not on device "
+                        f"{canonical_device!r} in the Live 12 source catalogue, "
+                        "and no close match was found. Kept as-is, but flagged "
+                        "as unverified."
+                    ),
+                    device=canonical_device,
+                    parameter=parameter,
+                    request_id=request_id,
+                )
+            )
+    else:
+        # Exact parameter match — the only case where a range check is sound.
+        spec = catalogue.parameter_spec(canonical_device, parameter)
+        if spec is not None and spec.has_range():
+            numeric_value = _coerce_value_to_number(record.get("value"))
+            if numeric_value is not None and not spec.in_range(numeric_value):
+                events.append(
+                    _build_unverified_event(
+                        path=base_path,
+                        reason=_REASON_VALUE_OUT_OF_RANGE,
+                        message=(
+                            f"Value {numeric_value} for {canonical_device!r}."
+                            f"{parameter!r} is outside the catalogue range "
+                            f"[{spec.min}, {spec.max}]. Kept as-is, but flagged "
+                            "as unverified."
+                        ),
+                        device=canonical_device,
+                        parameter=parameter,
+                        request_id=request_id,
+                        value=record.get("value"),
+                    )
+                )
+
+    if require_citation:
+        cited = record.get("phase1Fields")
+        non_empty = isinstance(cited, list) and any(
+            isinstance(s, str) and s.strip() for s in cited
+        )
+        if not non_empty:
+            events.append(
+                _build_unverified_event(
+                    path=f"{base_path}.phase1Fields",
+                    reason=_REASON_CITATION_MISSING,
+                    message=(
+                        "Recommendation has no Phase 1 citation. The chain of "
+                        "custody invariant requires every Gemini recommendation "
+                        "to cite at least one Phase 1 measurement field. Kept "
+                        "as-is, but flagged as unverified."
+                    ),
+                    device=canonical_device or device,
+                    parameter=parameter,
+                    request_id=request_id,
+                )
+            )
+
+
+def _apply_catalogue_checks_to_list(
+    *,
+    catalogue: Live12Catalogue,
+    phase2_result: dict[str, Any],
+    list_key: str,
+    base_path_prefix: str,
+    request_id: str,
+    require_citation: bool,
+    events: list[dict[str, Any]],
+) -> None:
+    items = phase2_result.get(list_key)
+    if not isinstance(items, list) or not items:
+        return
+    for index, item in enumerate(items):
+        record = _as_record(item)
+        if record is None:
+            # Opaque entries left untouched — shape salvage handled them already.
+            continue
+        _inspect_record(
+            catalogue=catalogue,
+            record=record,
+            base_path=f"{base_path_prefix}[{index}]",
+            request_id=request_id,
+            require_citation=require_citation,
+            events=events,
+        )
+
+
+def _apply_catalogue_checks_to_workflow_steps(
+    *,
+    catalogue: Live12Catalogue,
+    phase2_result: dict[str, Any],
+    request_id: str,
+    events: list[dict[str, Any]],
+) -> None:
+    secret_sauce = _as_record(phase2_result.get("secretSauce"))
+    if secret_sauce is None:
+        return
+    steps = secret_sauce.get("workflowSteps")
+    if not isinstance(steps, list) or not steps:
+        return
+    for index, item in enumerate(steps):
+        record = _as_record(item)
+        if record is None:
+            continue
+        # secretSauce workflow steps cite Phase 1 via `phase1Fields` too -- the
+        # schema requires the array, so the citation check applies.
+        _inspect_record(
+            catalogue=catalogue,
+            record=record,
+            base_path=f"secretSauce.workflowSteps[{index}]",
+            request_id=request_id,
+            require_citation=True,
+            events=events,
+        )
+
+
+def apply_live12_catalogue_gates(
+    phase2_result: dict[str, Any],
+    *,
+    request_id: str,
+    catalogue: Live12Catalogue | None = None,
+) -> list[dict[str, Any]]:
+    """Cross-check `phase2_result` against the Live 12 source catalogue and
+    return advisory warning events.
+
+    WARN-AND-KEEP: this function does NOT mutate `phase2_result`, does NOT drop
+    recommendations, and does NOT rewrite parameters. It only inspects and
+    returns warning-shaped dicts describing devices/parameters/citations the
+    catalogue cannot confirm. The caller stitches the events into the existing
+    `validationWarnings` channel. See the module docstring for why gating was
+    deliberately downgraded to warning.
+
+    The catalogue is loaded from `data/live12_catalogue.json` via
+    `Live12Catalogue.load_default()` when not injected explicitly (tests pass a
+    test catalogue here).
+    """
+    if not isinstance(phase2_result, dict):
+        return []
+    if catalogue is None:
+        catalogue = Live12Catalogue.load_default()
+    events: list[dict[str, Any]] = []
+    _apply_catalogue_checks_to_list(
+        catalogue=catalogue,
+        phase2_result=phase2_result,
+        list_key="mixAndMasterChain",
+        base_path_prefix="mixAndMasterChain",
+        request_id=request_id,
+        require_citation=True,
+        events=events,
+    )
+    _apply_catalogue_checks_to_list(
+        catalogue=catalogue,
+        phase2_result=phase2_result,
+        list_key="abletonRecommendations",
+        base_path_prefix="abletonRecommendations",
+        request_id=request_id,
+        require_citation=True,
+        events=events,
+    )
+    _apply_catalogue_checks_to_workflow_steps(
+        catalogue=catalogue,
+        phase2_result=phase2_result,
+        request_id=request_id,
+        events=events,
+    )
+    return events