feat(v7-h11): side_channels.apply RPC backend (per-family resolution + gotcha echo)

apstenku123 · apstenku123 · commit 6c6ac8fe889f · 2026-05-23T03:12:55.000+02:00
Closes the backend half of item 20 of the UI&lt;-&gt;API wiring honesty
list. Before this commit SideChannelsTab 'Apply' only wrote to local
spec — the UI could not learn whether a required family pointed at
columns the parquet shard didn't carry until the full verify ran.

side_channels.apply:
  - For each family computes active/inactive + columns_present /
    columns_missing + human-readable reason.
  - Reuses verify's _side_channel_policy_gotchas helper so required-
    column gotchas appear in both verify and apply.
  - Returns active_count / inactive_count so UI can render
    'applied: 3 of 5 families active' inline.

Honest scope: backend only — UI Apply-button wiring to call this RPC
ships in a follow-up commit.

bd: cppmega-mlx-wowq
7/7 pytest + 21/21 lifecycle+side-channel regression.
diff --git a/cppmega_v4/jsonrpc/dispatcher.py b/cppmega_v4/jsonrpc/dispatcher.py
@@ -58,6 +58,10 @@
     SideChannelPreviewParams,
     preview_side_channels,
 )
+from cppmega_v4.jsonrpc.side_channel_apply_method import (
+    SideChannelApplyParams,
+    apply_side_channels,
+)
 from cppmega_v4.jsonrpc.schema import (
     BuildPresetSpecsParams,
     CatalogExplainParams,
@@ -174,6 +178,10 @@
         SideChannelPreviewParams,
         lambda p, c: preview_side_channels(p, cache=c),
     ),
+    "side_channels.apply": (
+        SideChannelApplyParams,
+        lambda p, c: apply_side_channels(p, cache=c),
+    ),
 }
 
 
diff --git a/cppmega_v4/jsonrpc/side_channel_apply_method.py b/cppmega_v4/jsonrpc/side_channel_apply_method.py
@@ -0,0 +1,168 @@
+"""V7-H11: side_channels.apply RPC — verify a side-channel config.
+
+The Sidebar SideChannelsTab assembles a SideChannelSpecPayload (per-
+family mode/embedding/fallback + inference enrichment policy + a list
+of available_side_channels reported by the parquet shard) and asks
+the backend "is this config self-consistent, and if I plumb it into
+verify+train, what does it actually resolve to?".
+
+Before V7-H11 the Apply button only wrote to local spec; user had no
+backend confirmation that, e.g., a required family that asks for
+columns the parquet doesn't carry would fail at verify-time. This RPC
+runs the same gotcha-checker that verify does, plus echoes the set
+of families that would actually be active (mode != off + columns
+present in available_side_channels) so the UI can show "applied: 3 of
+5 families active".
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cppmega_v4.jsonrpc.cache import LRUCache, canonical_sha256
+from cppmega_v4.jsonrpc.schema import (
+    GotchaPayload, SideChannelSpecPayload,
+)
+
+
+class SideChannelApplyParams(BaseModel):
+    """Request: verify a side-channel config against available columns."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    side_channels: SideChannelSpecPayload
+    available_side_channels: list[str] = Field(
+        default_factory=lambda: ["doc_ids", "token_ids"],
+    )
+
+
+class FamilyApplyStatus(BaseModel):
+    """Per-family resolution: did this family actually engage?"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    family: str
+    mode: str
+    active: bool
+    reason: str
+    columns_requested: list[str]
+    columns_present: list[str]
+    columns_missing: list[str]
+
+
+class SideChannelApplyResult(BaseModel):
+    """V7-H11: side-channel apply verdict + per-family resolution."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    ok: bool
+    families: list[FamilyApplyStatus]
+    gotchas: list[GotchaPayload]
+    active_count: int
+    inactive_count: int
+    elapsed_ms: float
+
+
+def _resolve_family(
+    name: str, policy: Any, available: frozenset[str],
+    global_mode: str,
+) -> FamilyApplyStatus:
+    requested = list(policy.columns)
+    present = [c for c in requested if c in available]
+    missing = [c for c in requested if c not in available]
+    mode = policy.mode
+    effective_required = (mode == "require") or (global_mode == "require")
+    if mode == "off":
+        return FamilyApplyStatus(
+            family=name, mode=mode, active=False,
+            reason="family mode=off",
+            columns_requested=requested,
+            columns_present=present, columns_missing=missing,
+        )
+    if not requested:
+        return FamilyApplyStatus(
+            family=name, mode=mode, active=False,
+            reason="no columns declared for family",
+            columns_requested=requested,
+            columns_present=present, columns_missing=missing,
+        )
+    if not present:
+        reason = (
+            "required columns missing"
+            if effective_required else
+            "no requested columns present in shard"
+        )
+        return FamilyApplyStatus(
+            family=name, mode=mode, active=False, reason=reason,
+            columns_requested=requested,
+            columns_present=present, columns_missing=missing,
+        )
+    if missing and effective_required:
+        return FamilyApplyStatus(
+            family=name, mode=mode, active=False,
+            reason=f"required columns missing: {', '.join(missing)}",
+            columns_requested=requested,
+            columns_present=present, columns_missing=missing,
+        )
+    reason = (
+        "all requested columns present"
+        if not missing else
+        f"partial: {len(present)}/{len(requested)} columns present"
+    )
+    return FamilyApplyStatus(
+        family=name, mode=mode, active=True, reason=reason,
+        columns_requested=requested,
+        columns_present=present, columns_missing=missing,
+    )
+
+
+def apply_side_channels(
+    params: SideChannelApplyParams,
+    *,
+    cache: LRUCache | None = None,
+) -> SideChannelApplyResult:
+    """Verify side-channel config; report per-family resolution + gotchas."""
+    cache_key = "side-channel-apply::" + canonical_sha256(
+        params.model_dump(mode="json"),
+    )
+    if cache is not None:
+        hit = cache.get(cache_key)
+        if hit is not None:
+            return hit
+
+    t0 = time.perf_counter()
+    available = frozenset(params.available_side_channels)
+    global_mode = params.side_channels.mode
+    families: list[FamilyApplyStatus] = []
+    for name, policy in sorted(params.side_channels.families.items()):
+        families.append(_resolve_family(name, policy, available, global_mode))
+
+    # Reuse the same gotcha helper that verify uses so apply <-> verify
+    # agree on what's an error.
+    from cppmega_v4.jsonrpc.methods import _side_channel_policy_gotchas
+    gotchas = _side_channel_policy_gotchas(params.side_channels, available)
+
+    active = sum(1 for f in families if f.active)
+    inactive = sum(1 for f in families if not f.active)
+    ok = not any(g.severity == "error" for g in gotchas)
+
+    result = SideChannelApplyResult(
+        ok=ok,
+        families=families,
+        gotchas=gotchas,
+        active_count=active,
+        inactive_count=inactive,
+        elapsed_ms=round((time.perf_counter() - t0) * 1000.0, 3),
+    )
+    if cache is not None:
+        cache.set(cache_key, result)
+    return result
+
+
+__all__ = [
+    "SideChannelApplyParams", "SideChannelApplyResult",
+    "FamilyApplyStatus", "apply_side_channels",
+]
diff --git a/tests/v4/test_side_channels_apply.py b/tests/v4/test_side_channels_apply.py
@@ -0,0 +1,123 @@
+"""V7-H11: side_channels.apply RPC + per-family resolution + gotcha echo."""
+
+from __future__ import annotations
+
+from cppmega_v4.jsonrpc.side_channel_apply_method import (
+    SideChannelApplyParams, apply_side_channels,
+)
+from cppmega_v4.jsonrpc.schema import (
+    FamilySpecPayload, SideChannelSpecPayload,
+)
+
+
+def _spec_one(family_name: str, columns: list[str],
+              mode: str = "if_available") -> SideChannelSpecPayload:
+    return SideChannelSpecPayload(
+        mode="if_available",
+        families={
+            family_name: FamilySpecPayload(
+                mode=mode, columns=columns,
+                embedding="categorical", dropout=0.0,
+            ),
+        },
+    )
+
+
+def test_v7_h11_active_when_all_columns_present():
+    r = apply_side_channels(SideChannelApplyParams(
+        side_channels=_spec_one("platform", ["platform_ids"]),
+        available_side_channels=["platform_ids", "token_ids"],
+    ))
+    assert r.ok is True
+    assert r.active_count == 1
+    assert r.inactive_count == 0
+    assert r.families[0].family == "platform"
+    assert r.families[0].active is True
+    assert r.families[0].columns_missing == []
+    assert r.families[0].reason == "all requested columns present"
+
+
+def test_v7_h11_inactive_when_columns_absent_and_not_required():
+    r = apply_side_channels(SideChannelApplyParams(
+        side_channels=_spec_one("syntax",
+                                 ["token_ast_depth", "token_sibling_index"]),
+        available_side_channels=["doc_ids", "token_ids"],
+    ))
+    # Not required → gotcha NOT raised, but family is inactive.
+    assert r.ok is True
+    assert r.active_count == 0
+    assert r.inactive_count == 1
+    f = r.families[0]
+    assert f.active is False
+    assert f.columns_missing == ["token_ast_depth", "token_sibling_index"]
+    assert "no requested columns present" in f.reason
+
+
+def test_v7_h11_required_missing_columns_produce_gotcha_and_ok_false():
+    r = apply_side_channels(SideChannelApplyParams(
+        side_channels=_spec_one("platform", ["platform_ids"],
+                                 mode="require"),
+        available_side_channels=["doc_ids"],
+    ))
+    assert r.ok is False
+    assert any(g.id == "side_channel_required_platform"
+                for g in r.gotchas)
+    assert r.families[0].active is False
+
+
+def test_v7_h11_partial_columns_present_active_with_reason():
+    r = apply_side_channels(SideChannelApplyParams(
+        side_channels=_spec_one("structure",
+                                 ["token_structure_ids", "token_dep_levels",
+                                  "token_chunk_starts"]),
+        available_side_channels=["token_structure_ids", "doc_ids"],
+    ))
+    f = r.families[0]
+    assert f.active is True
+    assert set(f.columns_present) == {"token_structure_ids"}
+    assert set(f.columns_missing) == {"token_dep_levels",
+                                       "token_chunk_starts"}
+    assert "partial: 1/3" in f.reason
+
+
+def test_v7_h11_mode_off_yields_inactive_without_warning():
+    r = apply_side_channels(SideChannelApplyParams(
+        side_channels=_spec_one("temporal_diff", ["doc_ids"], mode="off"),
+        available_side_channels=["doc_ids"],
+    ))
+    assert r.ok is True
+    assert r.active_count == 0
+    assert r.families[0].reason == "family mode=off"
+    assert r.gotchas == []
+
+
+def test_v7_h11_dispatcher_routes_method_name():
+    """End-to-end: side_channels.apply method must reach the handler."""
+    from cppmega_v4.jsonrpc.dispatcher import dispatch
+    response = dispatch({
+        "jsonrpc": "2.0", "id": "T1", "method": "side_channels.apply",
+        "params": {
+            "side_channels": {
+                "mode": "if_available", "families": {},
+            },
+            "available_side_channels": ["doc_ids"],
+        },
+    })
+    assert response.error is None, response.error
+    assert response.result is not None
+    assert response.result["ok"] is True
+    assert response.result["families"] == []
+
+
+def test_v7_h11_dispatcher_rejects_unknown_extra_field():
+    """Schema is strict (extra=forbid)."""
+    from cppmega_v4.jsonrpc.dispatcher import dispatch
+    response = dispatch({
+        "jsonrpc": "2.0", "id": "T2", "method": "side_channels.apply",
+        "params": {
+            "side_channels": {"mode": "if_available", "families": {}},
+            "unknown_extra": 42,
+        },
+    })
+    assert response.error is not None
+    assert response.error.code == -32602  # INVALID_PARAMS
