refactor(wardline): centralize scan-routing validation in the service layer

The scan-route request-routing decision — "is request-side routing allowed,
and is the cell-spec well-formed?" — was hand-copied into both the HTTP
(/wardline/scan-results) and MCP (scan_route) adapters, alongside the cell-spec
parse and a byte-for-byte _parse_wardline_cell_map helper. The copies had
already drifted: HTTP rejected an empty cell_by_severity (422) while MCP
silently accepted an empty severity_map and routed nothing — the "check added
to one transport not the other" failure mode the layering exists to prevent.

Extract service.resolve_scan_routing (+ ResolvedRouting, _parse_cell_map_env)
as the single decision. It raises WardlineRoutingError carrying a `kind`
discriminator; each adapter maps kind to its own taxonomy — HTTP 500/403/422
(_WARDLINE_ROUTING_STATUS), MCP collapses all three to INVALID_CELL_SPEC (via
_service_error, before the generic ServiceError case). Both dead
_parse_wardline_cell_map copies are removed; env reads stay in the adapters.

Behavior-preserving for every pinned case (HTTP 403/"server-owned" + 422; MCP
INVALID_CELL_SPEC; the malformed-finding path still raises INVALID_ARGUMENT
from inside route_wardline_scan, kept distinct from routing errors). One
intended change closes the drift: an empty per-severity map is now rejected up
front on both transports — no silent governance skip.

TDD: tests/service/test_wardline.py (14 cases) + a new MCP regression
test_scan_route_rejects_empty_severity_map. Full suite 699 passed, mypy + ruff
clean, coverage floors hold, policy-boundary-check + governance-gate + SEI
oracle green.

rc4 review finding #4 (legis-604ddb8dd4).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tachyon-beep

CHANGELOG.md

@@ -212,6 +212,19 @@ listed as not-yet-built.
   `canonical.canonical_json` (whose `ensure_ascii=False` would change the signed
   bytes). Behavior-preserving — existing per-channel golden vectors unchanged,
   plus a new cross-channel anti-drift test. No change to signatures on the wire.
+- **Wardline scan-routing validation centralised in the service layer** — "is
+  request-side routing allowed, and is the cell-spec well-formed?" is a
+  governance decision that was hand-copied into both the HTTP
+  (`/wardline/scan-results`) and MCP (`scan_route`) adapters, along with the
+  cell-spec parse and a `_parse_wardline_cell_map` helper. The copies had already
+  drifted: the HTTP adapter rejected an empty `cell_by_severity` (422) while MCP
+  silently accepted an empty `severity_map` and routed nothing. The decision now
+  lives in `service.resolve_scan_routing`, raising a `WardlineRoutingError` whose
+  `kind` each adapter maps to its own taxonomy (HTTP 500/403/422 by kind; MCP
+  collapses to `INVALID_CELL_SPEC`) — so a new routing rule is added once and
+  cannot reach one transport but not the other. Behavior-preserving for every
+  pinned case; the one intended change closes the drift (an empty per-severity
+  map is now rejected up front on both transports — no silent governance skip).
 
 ### Fixed
 - **Ingest accepts realistic scans** — the over-strict Wardline ingest validator

src/legis/api/app.py

@@ -48,7 +48,12 @@
 from legis.governance.signoff_binding import bind_signoff_to_issue
 from legis.identity.entity_key import EntityKey
 from legis.identity.resolver import IdentityResolver
-from legis.service.errors import AuditIntegrityError, InvalidArgumentError, NotEnabledError
+from legis.service.errors import (
+    AuditIntegrityError,
+    InvalidArgumentError,
+    NotEnabledError,
+    WardlineRoutingError,
+)
 from legis.service.governance import compute_override_rate as _compute_override_rate
 from legis.service.governance import evaluate_policy as _evaluate_policy
 from legis.service.governance import request_signoff as _request_signoff
@@ -58,7 +63,10 @@
 from legis.service.governance import submit_override as _submit_override
 from legis.service.governance import submit_protected_override as _submit_protected_override
 from legis.service.governance import verified_records as _verified_records
-from legis.service.wardline import route_wardline_scan as _route_wardline_scan
+from legis.service.wardline import (
+    resolve_scan_routing,
+    route_wardline_scan as _route_wardline_scan,
+)
 from legis.policy.grammar import PolicyGrammar, default_grammar
 from legis.pulls.models import PullRequest, PullRequestState
 from legis.pulls.surface import PullSurface
@@ -67,7 +75,6 @@
     ScanOutcome,
     WardlineDirtyTreeError,
     WardlinePayloadError,
-    WardlineSeverity,
 )
 
 security = HTTPBearer(auto_error=False)
@@ -247,20 +254,13 @@ class CheckRunIn(BaseModel):
     finished_at: str | None = None
 
 
-def _parse_wardline_cell_map(raw: str) -> dict[WardlineSeverity, WardlineCellPolicy]:
-    mapping: dict[WardlineSeverity, WardlineCellPolicy] = {}
-    for part in raw.split(","):
-        if not part.strip():
-            continue
-        severity_raw, sep, cell_raw = part.partition("=")
-        if not sep:
-            raise ValueError("cell map entries must be SEVERITY=cell")
-        mapping[WardlineSeverity[severity_raw.strip()]] = WardlineCellPolicy(
-            cell_raw.strip()
-        )
-    if not mapping:
-        raise ValueError("cell map must not be empty")
-    return mapping
+# Wardline scan-routing rejections (raised by service.resolve_scan_routing) map
+# to HTTP status by kind; the MCP adapter collapses the same kinds to one code.
+_WARDLINE_ROUTING_STATUS = {
+    WardlineRoutingError.SERVER_MISCONFIGURED: 500,
+    WardlineRoutingError.SERVER_OWNED: 403,
+    WardlineRoutingError.MALFORMED: 422,
+}
 
 
 def _check_to_dict(run: CheckRun) -> dict:
@@ -772,72 +772,37 @@ def policy_evaluate(body: PolicyEvalIn, actor: str = Depends(verify_writer)) ->
 
     @app.post("/wardline/scan-results")
     def wardline_scan_results(body: ScanResultsIn, actor: str = Depends(verify_writer)) -> dict:
-        server_cell = os.environ.get("LEGIS_WARDLINE_CELL")
-        server_cell_by_severity = os.environ.get("LEGIS_WARDLINE_CELL_BY_SEVERITY")
-        if server_cell and server_cell_by_severity:
-            raise HTTPException(status_code=500, detail="server Wardline routing is misconfigured")
-        server_routing = server_cell is not None or server_cell_by_severity is not None
-        if server_routing and (
-            body.cell is not None or body.cell_by_severity is not None or body.fail_on is not None
-        ):
-            raise HTTPException(status_code=403, detail="Wardline routing is server-owned")
-        if not server_routing:
-            if os.environ.get("LEGIS_UNSAFE_WARDLINE_REQUEST_ROUTING") != "1":
-                raise HTTPException(
-                    status_code=403,
-                    detail="Wardline routing is server-owned; configure LEGIS_WARDLINE_CELL or LEGIS_WARDLINE_CELL_BY_SEVERITY",
-                )
-            if body.fail_on is not None:
-                if body.cell is None or body.cell_by_severity is not None:
-                    raise HTTPException(
-                        status_code=422,
-                        detail="fail_on routing requires cell and forbids cell_by_severity",
-                    )
-            elif (body.cell is None) == (body.cell_by_severity is None):
-                raise HTTPException(status_code=422,
-                                    detail="provide exactly one of cell or cell_by_severity")
-            if body.cell_by_severity is not None and not body.cell_by_severity:
-                raise HTTPException(status_code=422, detail="cell_by_severity must not be empty")
-
-        policy: WardlineCellPolicy | None = None
-        cell_map: dict[WardlineSeverity, WardlineCellPolicy] | None = None
-        fail_on: WardlineSeverity | None = None
         try:
-            if server_cell_by_severity is not None:
-                cell_map = _parse_wardline_cell_map(server_cell_by_severity)
-                cells = set(cell_map.values())
-            elif server_cell is not None:
-                policy = WardlineCellPolicy(server_cell)
-                cells = {policy}
-            elif body.cell_by_severity is not None:
-                cell_map = {WardlineSeverity[sev]: WardlineCellPolicy(cell)
-                            for sev, cell in body.cell_by_severity.items()}
-                cells = set(cell_map.values())
-            else:
-                policy = WardlineCellPolicy(body.cell)
-                if body.fail_on is not None:
-                    fail_on = WardlineSeverity[body.fail_on]
-                    cells = {policy, WardlineCellPolicy.SURFACE_ONLY}
-                else:
-                    cells = {policy}
-        except (KeyError, ValueError) as exc:
-            raise HTTPException(status_code=422, detail=f"unknown cell/severity: {exc}")
+            routing = resolve_scan_routing(
+                server_cell=os.environ.get("LEGIS_WARDLINE_CELL"),
+                server_cell_by_severity=os.environ.get("LEGIS_WARDLINE_CELL_BY_SEVERITY"),
+                request_cell=body.cell,
+                request_severity_map=body.cell_by_severity,
+                request_fail_on=body.fail_on,
+                allow_request_routing=(
+                    os.environ.get("LEGIS_UNSAFE_WARDLINE_REQUEST_ROUTING") == "1"
+                ),
+            )
+        except WardlineRoutingError as exc:
+            raise HTTPException(
+                status_code=_WARDLINE_ROUTING_STATUS[exc.kind], detail=str(exc)
+            ) from exc
 
         # Only provision the governance store when a surface cell can actually run:
         # engine() lazily creates .weft/legis/legis-governance.db, so a pure block_escalate scan
         # must not touch it. signoff_gate is an injected param (no side effect).
-        needs_engine = bool(cells & {WardlineCellPolicy.SURFACE_OVERRIDE,
-                                     WardlineCellPolicy.SURFACE_ONLY})
+        needs_engine = bool(routing.cells & {WardlineCellPolicy.SURFACE_OVERRIDE,
+                                             WardlineCellPolicy.SURFACE_ONLY})
         try:
             routed = _route_wardline_scan(
                 body.scan,
                 agent_id=_recorded_actor(actor, body.agent_id),
                 identity=identity,
                 engine=engine() if needs_engine else None,
                 signoff=signoff_gate,
-                policy=policy,
-                cell_map=cell_map,
-                fail_on=fail_on,
+                policy=routing.policy,
+                cell_map=routing.cell_map,
+                fail_on=routing.fail_on,
                 artifact_key=(
                     os.environ["LEGIS_WARDLINE_ARTIFACT_KEY"].encode("utf-8")
                     if os.environ.get("LEGIS_WARDLINE_ARTIFACT_KEY")

src/legis/mcp.py

@@ -43,6 +43,7 @@
     NotEnabledError,
     NotFoundError,
     ServiceError,
+    WardlineRoutingError,
 )
 from legis.service.explain import explain_policy
 from legis.service.governance import (
@@ -53,10 +54,9 @@
     request_signoff,
     verified_records as service_verified_records,
 )
-from legis.service.wardline import route_wardline_scan
+from legis.service.wardline import resolve_scan_routing, route_wardline_scan
 from legis.store.audit_store import AuditStore
-from legis.wardline.governor import WardlineCellPolicy
-from legis.wardline.ingest import ScanOutcome, WardlineDirtyTreeError, WardlineSeverity
+from legis.wardline.ingest import ScanOutcome, WardlineDirtyTreeError
 
 
 _AGENT_TOOLS = frozenset(
@@ -414,6 +414,11 @@ def _service_error(exc: Exception) -> dict[str, Any]:
         return _tool_error("NOT_FOUND", str(exc))
     if isinstance(exc, InvalidArgumentError):
         return _tool_error("INVALID_ARGUMENT", str(exc))
+    if isinstance(exc, WardlineRoutingError):
+        # All three routing kinds (server-misconfigured / server-owned /
+        # malformed) collapse to one MCP code; the HTTP adapter splits them by
+        # status. Must precede the generic ServiceError case below.
+        return _tool_error("INVALID_CELL_SPEC", str(exc))
     if isinstance(exc, GitError):
         return _tool_error("GIT_ERROR", str(exc))
     if isinstance(exc, ServiceError):
@@ -519,22 +524,6 @@ def _registry(runtime: McpRuntime) -> PolicyCellRegistry:
     return runtime.cell_registry or fail_closed_policy_cells()
 
 
-def _parse_wardline_cell_map(raw: str) -> dict[WardlineSeverity, WardlineCellPolicy]:
-    mapping: dict[WardlineSeverity, WardlineCellPolicy] = {}
-    for part in raw.split(","):
-        if not part.strip():
-            continue
-        severity_raw, sep, cell_raw = part.partition("=")
-        if not sep:
-            raise ValueError("cell map entries must be SEVERITY=cell")
-        mapping[WardlineSeverity[severity_raw.strip()]] = WardlineCellPolicy(
-            cell_raw.strip()
-        )
-    if not mapping:
-        raise ValueError("cell map must not be empty")
-    return mapping
-
-
 def _explanation_payload(explanation) -> dict[str, Any]:
     payload = explanation.to_payload()
     payload["available_moves"] = [
@@ -925,69 +914,34 @@ def _tool_policy_evaluate(runtime: McpRuntime, args: dict[str, Any]) -> dict[str
 
 
 def _tool_scan_route(runtime: McpRuntime, args: dict[str, Any]) -> dict[str, Any]:
-    server_cell = os.environ.get("LEGIS_WARDLINE_CELL")
-    server_cell_by_severity = os.environ.get("LEGIS_WARDLINE_CELL_BY_SEVERITY")
-    if server_cell and server_cell_by_severity:
-        return _tool_error(
-            "INVALID_CELL_SPEC", "server Wardline routing is misconfigured"
-        )
-    has_cell = "cell" in args
-    has_map = "severity_map" in args
-    has_fail_on = "fail_on" in args
-    server_routing = server_cell is not None or server_cell_by_severity is not None
-    if server_routing and (has_cell or has_map or has_fail_on):
-        return _tool_error(
-            "INVALID_CELL_SPEC", "Wardline routing is server-owned"
-        )
-    if not server_routing:
-        if os.environ.get("LEGIS_UNSAFE_WARDLINE_REQUEST_ROUTING") != "1":
-            return _tool_error(
-                "INVALID_CELL_SPEC",
-                "Wardline routing is server-owned; configure "
-                "LEGIS_WARDLINE_CELL or LEGIS_WARDLINE_CELL_BY_SEVERITY",
-            )
-        if has_fail_on:
-            if not has_cell or has_map:
-                return _tool_error(
-                    "INVALID_CELL_SPEC",
-                    "fail_on routing requires cell and forbids severity_map",
-                )
-        elif has_cell == has_map:
-            return _tool_error(
-                "INVALID_CELL_SPEC",
-                "provide exactly one of cell or severity_map",
-            )
+    # "severity_map" must be an object if present (transport-type check); the
+    # governance decision — is request routing allowed, and is the spec
+    # well-formed? — lives in resolve_scan_routing, shared with the HTTP adapter.
+    # A WardlineRoutingError propagates to call_tool's translator → INVALID_CELL_SPEC.
+    request_severity_map = (
+        _require_object(args, "severity_map") if "severity_map" in args else None
+    )
+    routing = resolve_scan_routing(
+        server_cell=os.environ.get("LEGIS_WARDLINE_CELL"),
+        server_cell_by_severity=os.environ.get("LEGIS_WARDLINE_CELL_BY_SEVERITY"),
+        request_cell=args.get("cell"),
+        request_severity_map=request_severity_map,
+        request_fail_on=args.get("fail_on"),
+        allow_request_routing=(
+            os.environ.get("LEGIS_UNSAFE_WARDLINE_REQUEST_ROUTING") == "1"
+        ),
+    )
     scan = _require_object(args, "scan")
-    scan_policy: WardlineCellPolicy | None = None
-    scan_cell_map: dict[WardlineSeverity, WardlineCellPolicy] | None = None
-    scan_fail_on: WardlineSeverity | None = None
-    try:
-        if server_cell_by_severity is not None:
-            scan_cell_map = _parse_wardline_cell_map(server_cell_by_severity)
-        elif server_cell is not None:
-            scan_policy = WardlineCellPolicy(server_cell)
-        elif has_cell:
-            scan_policy = WardlineCellPolicy(_require(args, "cell"))
-            if has_fail_on:
-                scan_fail_on = WardlineSeverity[_require(args, "fail_on")]
-        else:
-            raw_map = _require_object(args, "severity_map")
-            scan_cell_map = {
-                WardlineSeverity[severity]: WardlineCellPolicy(cell)
-                for severity, cell in raw_map.items()
-            }
-    except (KeyError, ValueError) as exc:
-        return _tool_error("INVALID_CELL_SPEC", str(exc))
     try:
         routed = route_wardline_scan(
             scan,
             agent_id=runtime.agent_id,
             identity=runtime.identity,
             engine=_engine(runtime),
             signoff=runtime.signoff_gate,
-            policy=scan_policy,
-            cell_map=scan_cell_map,
-            fail_on=scan_fail_on,
+            policy=routing.policy,
+            cell_map=routing.cell_map,
+            fail_on=routing.fail_on,
             artifact_key=(
                 runtime.wardline_artifact_key
                 or (

src/legis/service/errors.py

@@ -28,6 +28,25 @@ class InvalidArgumentError(ServiceError):
     """Caller input is structurally valid for the transport but invalid for Legis."""
 
 
+class WardlineRoutingError(ServiceError):
+    """A Wardline scan-routing request is not permitted or is malformed.
+
+    Carries a ``kind`` discriminator so each adapter can preserve its own
+    taxonomy without re-implementing the decision: the HTTP adapter maps
+    ``server_misconfigured`` → 500, ``server_owned`` → 403, ``malformed`` → 422,
+    while the MCP adapter collapses all three to ``INVALID_CELL_SPEC``. Adapters
+    switch on the ``kind`` attribute, never on message text.
+    """
+
+    SERVER_MISCONFIGURED = "server_misconfigured"
+    SERVER_OWNED = "server_owned"
+    MALFORMED = "malformed"
+
+    def __init__(self, kind: str, message: str) -> None:
+        super().__init__(message)
+        self.kind = kind
+
+
 class ProtectedKeyRequiredError(ServiceError):
     """A protected trail was read without the HMAC key needed to verify it.