@@ -27,6 +27,11 @@ export TRACE_STORE_ROOT=/srv/copilot/traces
 export ECO_MODEL=YOUR_PINNED_GENERATOR_MODEL
 export PERFORMANCE_MODEL=YOUR_PINNED_INVESTIGATOR_MODEL
 export IMPROVE_JUDGE=api:YOUR_PINNED_JUDGE_MODEL
+# 判官也可走订阅 CLI：cli:<cursor|codex|claude|zcode>:<model>。cli:zcode 经生产
+# ZCodeTransport 的 tool-less 一次性调用执行（与 kb 质量门同一工具：空 scratch 目录、
+# 移除全部原生工具、容器审计、served-model 断言；推理档位取环境 ZCODE_REASONING_LEVEL），
+# 订阅计费只计入判官调用次数预算，USD 记 0。
+# export IMPROVE_JUDGE=cli:zcode:GLM-5.3
 
 infermatrix-copilot improve workflows list
 infermatrix-copilot improve workflows check --workflow pr-review.agent.review_diff
 
@@ -1,18 +1,25 @@
 """Judge calls, all under the engine's ledger (design §10).
 
 The engine never starts a judge subprocess directly; every judge call goes
-through `run_judge`, which has two paths:
+through `run_judge`, which has three paths:
 
 * **API judges** through `LLM.create` — captured as `model_call` records by
   the choke point and priced against the dollar envelope (P3 governor);
 * **CLI judges** (a subscription harness such as `cursor-agent`) through
   `governed_subprocess` — one call reserved against the judge-call envelope,
   tool-less by construction (fresh empty workspace, `--mode ask`, any tool
   call in the stream fails the verdict), recorded as a `model_call` with
-  ``usd=0``.
-
-A `Governor` (P3) gates both; without one the calls are unmetered but still
-recorded. `parse_verdict` extracts the first JSON object of a reply.
+  ``usd=0``;
+* **zcode judges** (``cli:zcode:<model>``) through the production
+  `ZCodeTransport.complete` one-shot — the same tool-less instrument the
+  knowledge gate's `ModelGateway` uses (empty scratch cwd, every native
+  tool removed, containment audit, served-model assertion; the reasoning
+  level is the ambient `ZCODE_REASONING_LEVEL`). Reserved and settled
+  against the judge-call envelope like any CLI judge; subscription billing,
+  so ``usd=0`` and only the call count is governed.
+
+A `Governor` (P3) gates all of them; without one the calls are unmetered but
+still recorded. `parse_verdict` extracts the first JSON object of a reply.
 """
 
 from __future__ import annotations
@@ -38,7 +45,7 @@ class JudgeError(RuntimeError):
 class JudgeSpec:
     kind: str                 # "api" | "cli"
     model: str
-    provider: str = ""        # cli: cursor | codex | claude ; api: the LLM's provider
+    provider: str = ""        # cli: cursor | codex | claude | zcode ; api: the LLM's provider
     max_tokens: int = 2000
 
 
@@ -153,10 +160,38 @@ def governed_subprocess(argv: list[str], *, governor: Any = None, timeout: int =
             governor.settle_judge_call(token)
 
 
+def _run_zcode(spec: JudgeSpec, *, system: str, prompt: str, role: str,
+               governor: Any) -> tuple[str, dict, str, str]:
+    """One governed zcode judge call through the production transport: the
+    same tool-less one-shot the kb gate's ``ModelGateway`` uses (empty
+    scratch cwd, every native tool removed, containment audit, served-model
+    assertion), so an experiment's gate judge IS the production instrument,
+    not an API proxy of it. Subscription billing: ``usd`` stays 0 and only
+    the call count draws on the judge-call envelope."""
+    from ..config import Settings
+    from ..providers.registry import transport_for_id
+
+    token = governor.reserve_judge_call() if governor is not None else None
+    try:
+        transport = transport_for_id(Settings(), "zcode")
+        reply = transport.complete(system=system,
+                                   messages=[{"role": "user", "content": prompt}],
+                                   model=spec.model, role=role)
+        text = "".join(b.text for b in reply.blocks if b.type == "text")
+        usage = dict(getattr(reply, "usage", None) or {})
+        return text, usage, "" if text else "empty zcode judge reply", \
+            str(getattr(reply, "model", "") or "")
+    except Exception as exc:  # noqa: BLE001 — the verdict records the reason
+        return "", {}, f"zcode judge unavailable: {type(exc).__name__}: {exc}", ""
+    finally:
+        if governor is not None:
+            governor.settle_judge_call(token)
+
+
 def run_judge(spec: JudgeSpec, *, system: str, prompt: str, llm: Any = None, governor: Any = None,
               role: str = "judge", runner: Callable[..., Any] | None = None) -> dict:
     """Run one judge call and return its parsed JSON verdict (raises
-    JudgeError on an unusable reply). Both paths are recorded as
+    JudgeError on an unusable reply). All paths are recorded as
     ``model_call`` records under the current trace context."""
     if spec.kind == "api":
         if llm is None:
@@ -167,25 +202,31 @@ def run_judge(spec: JudgeSpec, *, system: str, prompt: str, llm: Any = None, gov
         return parse_verdict(text)
     if spec.kind != "cli":
         raise JudgeError(f"unknown judge kind {spec.kind!r}")
-    workspace = Path(tempfile.mkdtemp(prefix="judge-ws-"))
-    workspace.chmod(0o700)
     started = time.monotonic()
-    try:
-        full = prompt if not system else f"{system}\n\n{prompt}"
-        proc = governed_subprocess(_cli_argv(spec, full, workspace), governor=governor, cwd=str(workspace),
-                                   runner=runner, input=full if spec.provider == "codex" else None)
-        text, usage = _cli_result(spec, getattr(proc, "stdout", "") or "")
-        error = "" if text else f"empty judge reply (rc={getattr(proc, 'returncode', '?')})"
-    except JudgeError as exc:
-        text, usage, error = "", {}, str(exc)
-    finally:
-        shutil.rmtree(workspace, ignore_errors=True)
+    if spec.provider == "zcode":
+        text, usage, error, served = _run_zcode(spec, system=system, prompt=prompt,
+                                                role=role, governor=governor)
+    else:
+        workspace = Path(tempfile.mkdtemp(prefix="judge-ws-"))
+        workspace.chmod(0o700)
+        try:
+            full = prompt if not system else f"{system}\n\n{prompt}"
+            proc = governed_subprocess(_cli_argv(spec, full, workspace), governor=governor, cwd=str(workspace),
+                                       runner=runner, input=full if spec.provider == "codex" else None)
+            text, usage = _cli_result(spec, getattr(proc, "stdout", "") or "")
+            error = "" if text else f"empty judge reply (rc={getattr(proc, 'returncode', '?')})"
+        except JudgeError as exc:
+            text, usage, error = "", {}, str(exc)
+        finally:
+            shutil.rmtree(workspace, ignore_errors=True)
+        served = ""
     store = current_store()
     if store is not None:
         try:
             store.append("model_call", inputs={"system": system, "prompt": prompt},
                          outputs={"reply": text} if text else {},
-                         model={"role": role, "provider": spec.provider, "model": spec.model, "served_model": ""},
+                         model={"role": role, "provider": spec.provider, "model": spec.model,
+                                "served_model": served},
                          usage=usage, seconds=round(time.monotonic() - started, 3),
                          result={"format": "prompt/1", "usd": 0.0, "subscription": True}, error=error)
         except Exception:  # noqa: BLE001 - recording never changes a verdict
@@ -197,14 +238,16 @@ def run_judge(spec: JudgeSpec, *, system: str, prompt: str, llm: Any = None, gov
 
 def judge_spec_from(settings: Any, override: str = "") -> JudgeSpec | None:
     """The gold_match / paired judge from ``settings.improve_judge`` (or an
-    explicit override): ``api:<model>`` or ``cli:<provider>:<model>``; None
-    when unset."""
+    explicit override): ``api:<model>`` or ``cli:<provider>:<model>`` where
+    the provider is cursor | codex | claude | zcode; None when unset."""
     raw = str(override or getattr(settings, "improve_judge", "") or "")
     if not raw:
         return None
     parts = raw.split(":")
     if parts[0] == "api" and len(parts) == 2:
         return JudgeSpec("api", parts[1])
     if parts[0] == "cli" and len(parts) == 3:
+        if parts[1] not in ("cursor", "codex", "claude", "zcode"):
+            raise JudgeError(f"unknown CLI judge provider {parts[1]!r}")
         return JudgeSpec("cli", parts[2], provider=parts[1])
     raise JudgeError(f"improve_judge must be api:<model> or cli:<provider>:<model>, got {raw!r}")
@@ -136,6 +136,138 @@ def create(self, **kw):
         run_judge(JudgeSpec("api", "m"), system="s", prompt="p")
 
 
+def test_zcode_judge_spec_parsing():
+    from infermatrix_copilot.improve.judges import judge_spec_from
+
+    spec = judge_spec_from(SimpleNamespace(improve_judge="cli:zcode:GLM-5.3"))
+    assert spec == JudgeSpec("cli", "GLM-5.3", provider="zcode")
+    assert judge_spec_from(SimpleNamespace(improve_judge="")) is None
+    with pytest.raises(JudgeError, match="unknown CLI judge provider"):
+        judge_spec_from(SimpleNamespace(improve_judge="cli:cursor-agent:m"))
+    with pytest.raises(JudgeError, match="improve_judge must be"):
+        judge_spec_from(SimpleNamespace(improve_judge="zcode:GLM-5.3"))
+
+
+def test_zcode_judge_delegates_to_the_production_transport(tmp_path, monkeypatch):
+    """cli:zcode judging goes through ZCodeTransport.complete — the same
+    tool-less one-shot the production kb gate uses — governed by the
+    judge-call envelope and recorded like any CLI judge."""
+    from infermatrix_copilot.improve.judges import _run_zcode
+    from infermatrix_copilot.providers import registry
+
+    calls = []
+
+    class Gov:
+        def reserve_judge_call(self):
+            calls.append("reserve")
+            return "2027-W03|judge|z"
+
+        def settle_judge_call(self, token=None):
+            calls.append(("settle", token))
+
+    class Transport:
+        def __init__(self, reply):
+            self.reply = reply
+            self.kwargs = None
+
+        def complete(self, **kw):
+            self.kwargs = kw
+            if isinstance(self.reply, Exception):
+                raise self.reply
+            return self.reply
+
+    ok = SimpleNamespace(blocks=[SimpleNamespace(type="text", text='{"status": "hit", "quote": "q"}')],
+                         usage={"input_tokens": 7, "output_tokens": 3}, model="GLM-5.3")
+    transport = Transport(ok)
+    monkeypatch.setattr(registry, "transport_for_id", lambda settings, pid: transport)
+
+    store = TraceStore(tmp_path / "t", environ={})
+    with bind_store(store), trace_context(run_id="jz", unit_id="uz"):
+        verdict = run_judge(JudgeSpec("cli", "GLM-5.3", provider="zcode"), system="s", prompt="p",
+                            governor=Gov())
+    assert verdict["status"] == "hit"
+    assert transport.kwargs == {"system": "s", "messages": [{"role": "user", "content": "p"}],
+                                "model": "GLM-5.3", "role": "judge"}
+    assert calls == ["reserve", ("settle", "2027-W03|judge|z")]
+    rec = store.query(kind="model_call", unit_id="uz")[0]
+    assert rec["model"]["provider"] == "zcode" and rec["model"]["served_model"] == "GLM-5.3"
+    assert rec["result"]["usd"] == 0.0 and rec["result"]["subscription"] is True
+    assert rec["usage"]["input_tokens"] == 7 and not rec["error"]
+
+    down = Transport(RuntimeError("zcode CLI is not installed"))
+    monkeypatch.setattr(registry, "transport_for_id", lambda settings, pid: down)
+    with pytest.raises(JudgeError, match="zcode judge unavailable"):
+        run_judge(JudgeSpec("cli", "GLM-5.3", provider="zcode"), system="s", prompt="p", governor=Gov())
+    assert calls[-1] == ("settle", "2027-W03|judge|z")   # a failed call still settles
+
+    empty = Transport(SimpleNamespace(blocks=[], usage={}, model="GLM-5.3"))
+    monkeypatch.setattr(registry, "transport_for_id", lambda settings, pid: empty)
+    with pytest.raises(JudgeError, match="empty zcode judge reply"):
+        run_judge(JudgeSpec("cli", "GLM-5.3", provider="zcode"), system="s", prompt="p", governor=None)
+
+
+def test_zcode_run_helper_returns_the_parsed_triple(monkeypatch):
+    """The helper's contract: (text, usage, error, served) — the reply's
+    text from text blocks only, non-text blocks ignored."""
+    from infermatrix_copilot.improve.judges import _run_zcode
+    from infermatrix_copilot.providers import registry
+
+    class Transport:
+        def complete(self, **kw):
+            return SimpleNamespace(blocks=[SimpleNamespace(type="reasoning", text="thinking"),
+                                           SimpleNamespace(type="text", text='{"verdict": "consistent"}')],
+                                   usage={"input_tokens": 1}, model="GLM-5.3")
+
+    monkeypatch.setattr(registry, "transport_for_id", lambda settings, pid: Transport())
+    text, usage, error, served = _run_zcode(JudgeSpec("cli", "GLM-5.3", provider="zcode"),
+                                            system="s", prompt="p", role="judge", governor=None)
+    assert text == '{"verdict": "consistent"}' and not error and served == "GLM-5.3"
+    assert usage == {"input_tokens": 1}
+
+
+def test_zcode_judge_serves_the_trusted_kb_gate(monkeypatch, tmp_path):
+    """End to end: the trusted driver's Gateway over a cli:zcode judge spec
+    drives the production gate (judge_block + consistency) — the
+    experiment-internal gate judge is the production instrument, with the
+    ModelRole the gate records naming the zcode backend."""
+    import json as _json
+
+    from infermatrix_copilot.improve.drivers import Gateway
+    from infermatrix_copilot.improve.judges import judge_spec_from
+    from infermatrix_copilot.kb_service import gate as kb_gate
+    from infermatrix_copilot.kb_service.models import ModelRole
+    from infermatrix_copilot.providers import registry
+    from test_kb_intake_gate import PAGE, Op, _rule, _tree
+
+    class Transport:
+        def complete(self, *, system, messages, model, role, **kw):
+            prompt = messages[0]["content"]
+            if '"directory"' in prompt:                      # the consistency call
+                return SimpleNamespace(blocks=[SimpleNamespace(type="text", text='{"verdict": "consistent", "conflicts": []}')],
+                                       usage={}, model=model)
+            payload = _json.loads(prompt.split("<untrusted_data>\n", 1)[1].rsplit("\n</untrusted_data>", 1)[0])
+            dims = {d: "yes" for d in payload["dimensions_to_answer"]}
+            return SimpleNamespace(blocks=[SimpleNamespace(type="text", text=_json.dumps(
+                {"dimensions": dims, "reasons": {}}))], usage={}, model=model)
+
+    monkeypatch.setattr(registry, "transport_for_id", lambda settings, pid: Transport())
+    spec = judge_spec_from(SimpleNamespace(improve_judge="cli:zcode:GLM-5.3"))
+    gateway = Gateway(None, spec, None)
+    judge = ModelRole("judge", spec.provider, spec.model)
+
+    from infermatrix_copilot.kb_service.gate import changes_between
+    from infermatrix_copilot.knowledge_service.ops import apply_operations
+
+    base = _tree()
+    add = [Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a", "PR #11"))]
+    head = {**base, **apply_operations(base, add, release="v1", today="2026-09-28").files}
+    decision = kb_gate.run_gate(base=base, head=head, changes=changes_between(base, head),
+                                external_texts={}, evidence=[{"source_reference": "PR #11"}],
+                                gateway=gateway, judge=judge, release="v1", repo_dir="repos/demo")
+    assert decision.status == "pass"
+    assert all(b.model == "GLM-5.3" for b in decision.blocks)
+
+
 def test_paired_statistics_labels_power_and_kappa():
     r = stats.paired({f"i{k}": [0.1, 0.12] for k in range(10)}, "recall")
     assert r.n_items == 10 and r.n_verdicts == 20 and abs(r.mean - 0.11) < 1e-9 and r.lo > 0
 
@@ -183,25 +183,27 @@ def test_gate_with_a_zcode_judge_uses_the_same_instrument():
 def test_judge_tuning_is_the_single_source_the_gate_uses():
     """judge_tuning is the evolvable surface; gate.py must delegate to it and
     stay byte-identical to the pre-extraction baseline (evolution starts
-    from the production behaviour, never from a silent rewording)."""
-    import subprocess
+    from the production behaviour, never from a silent rewording). The
+    digests pin the shipped baseline — recorded when the module was
+    extracted from gate.py (7aa922dd9^)."""
+    import hashlib
 
     from infermatrix_copilot.kb_service import gate, judge_tuning
 
     assert gate.JUDGE_SYSTEM is judge_tuning.JUDGE_SYSTEM
     assert gate.CONSISTENCY_SYSTEM is judge_tuning.CONSISTENCY_SYSTEM
     assert gate.NEIGHBOUR_LIMIT is judge_tuning.NEIGHBOUR_LIMIT == 30
 
-    old = subprocess.run(["git", "show", "HEAD:src/infermatrix_copilot/kb_service/gate.py"],
-                         capture_output=True, text=True, check=True,
-                         cwd=Path(__file__).resolve().parents[1]).stdout
-    import re
-
-    def extract(name):
-        m = re.search(rf'^{name} = (""".*?""")$', old, re.M | re.S)
-        return eval(m.group(1))  # noqa: S307 - test fixture over our own constant
-    assert judge_tuning.JUDGE_SYSTEM == extract("JUDGE_SYSTEM")
-    assert judge_tuning.CONSISTENCY_SYSTEM == extract("CONSISTENCY_SYSTEM")
+    baseline = {
+        "JUDGE_SYSTEM": "2d120ade58b9ac32c0af4b0ad8854205f80fbfcad4e89ff2204b47028ec0781d",
+        "CONSISTENCY_SYSTEM": "db4fd2396f0400f9d45311f0c1460d181d83f8b64489c142c398bd1eca3f0b97",
+    }
+    for name, expected in baseline.items():
+        actual = hashlib.sha256(getattr(judge_tuning, name).encode()).hexdigest()
+        assert actual == expected, (
+            f"{name} drifted from the production baseline the evolution "
+            "engine starts from; rewording must arrive as an engine candidate, "
+            "not as part of another change")
 
 
 def test_verdict_aggregation_keeps_the_fail_closed_ordering():