fix(demos): demo 10 denied every call on a catalog drift CI could not see

.github/workflows/ci.yml

@@ -41,6 +41,14 @@ jobs:
           test -f web-console/vendor/examples/financial-services/policy/allow.cedar
           python -m py_compile web-console/run.py web-console/webserver.py web-console/policy_variants.py
 
+      # Static checks first: they need no ports and fail in under a second.
+      # test_catalog_definition_drift catches a catalog that no longer matches
+      # its upstream server, which the gateway treats as a rug pull and denies.
+      - name: Tests
+        run: |
+          pip install pytest
+          python -m pytest tests/ -q
+
       # Run in order: demo-01 produces workspace/trace-claim.json, which
       # demo-02 and demo-03 consume.
       - name: Demo 1 - cMCP in action

demo-10-model-gateway/catalog.json

@@ -46,13 +46,25 @@
       "output_schema": {
         "type": "object",
         "properties": {
-          "result": {
+          "model": {
+            "type": "string"
+          },
+          "region": {
+            "type": "string"
+          },
+          "cloud": {
+            "type": "string"
+          },
+          "content": {
+            "type": "string"
+          },
+          "upstream": {
             "type": "string"
           }
         }
       }
     },
-    "definition_hash": "sha256:f59f39e06537b04c1acf7fa975b3f8a0367f3a468bf600b70bda38318c9f283c",
+    "definition_hash": "sha256:a1df74a4a40c99c90cdd565e0093e107aec863efc47a135e1a9940b2d7ff9384",
     "compliance_domain": "internal",
     "requires_baa": false,
     "sensitivity_level": "confidential",

demo-10-model-gateway/client.py

@@ -7,6 +7,7 @@
 """
 import json
 import sys
+import urllib.error
 import urllib.request
 
 from openai import OpenAI
@@ -74,8 +75,16 @@ def main():
 
     print(BAR)
     print("\nClosing the session and fetching the signed claim...\n")
-    with urllib.request.urlopen(f"{GATEWAY}/trust-record", timeout=20) as r:
-        record = json.loads(r.read())
+    try:
+        with urllib.request.urlopen(f"{GATEWAY}/trust-record", timeout=20) as r:
+            record = json.loads(r.read())
+    except urllib.error.HTTPError as exc:
+        detail = exc.read().decode(errors="replace")[:300]
+        print(f"  could not fetch the signed claim: HTTP {exc.code} {detail}")
+        print("  Nothing to sign means no call was allowed. Check model-gateway.log for")
+        print("  UPSTREAM_CATALOG_DRIFT: catalog.json and model_server.py must describe")
+        print("  the same tool, or the gateway fail-closes on every call.")
+        return 1
     trace = record.get("trace", {})
     gw = record.get("gateway", {})
     print(f"  policy.bundle_hash   {trace.get('policy', {}).get('bundle_hash', '')}")

demo-10-model-gateway/model_server.py

@@ -68,6 +68,10 @@ def _err(id_, code: int, msg: str) -> dict:
     return {"jsonrpc": "2.0", "id": id_, "error": {"code": code, "message": msg}}
 
 
+# The catalog's approved_definition is the contract. cMCP hashes description +
+# input schema + output schema on both sides and fail-closes on any difference,
+# so this dict and catalog.json's approved_definition must stay identical: an
+# argument the endpoint sends but this schema omits reads as a rug pull.
 TOOL = {
     "name": "model.chat_completion",
     "description": "Run a chat completion against a model in the catalog",
@@ -80,10 +84,21 @@ def _err(id_, code: int, msg: str) -> dict:
             "model_region": {"type": "string"},
             "model_cloud": {"type": "string"},
             "data_class": {"type": "string"},
+            "contains_identifiers": {"type": "boolean"},
             "redacted": {"type": "boolean"},
             "redaction_count": {"type": "number"},
         },
     },
+    "outputSchema": {
+        "type": "object",
+        "properties": {
+            "model": {"type": "string"},
+            "region": {"type": "string"},
+            "cloud": {"type": "string"},
+            "content": {"type": "string"},
+            "upstream": {"type": "string"},
+        },
+    },
 }
 
 

demo-10-model-gateway/run.py

@@ -93,7 +93,12 @@ def main():
             while True:
                 time.sleep(1)
         else:
-            subprocess.call([sys.executable, str(HERE / "client.py")], cwd=HERE)
+            # The client's exit code is this demo's exit code. Discarding it kept
+            # CI green through a gateway that denied every call.
+            rc = subprocess.call([sys.executable, str(HERE / "client.py")], cwd=HERE)
+            if rc != 0:
+                print(f"client.py exited with code {rc}. See the *.log files.", file=sys.stderr)
+                return rc
     except KeyboardInterrupt:
         pass
     finally:

tests/test_catalog_definition_drift.py

@@ -0,0 +1,94 @@
+"""Static guard against the drift that broke demo 10.
+
+cMCP hashes description + input schema + output schema on both sides of the
+boundary and fail-closes when they differ. A catalog entry that describes an
+argument the upstream server does not advertise is a `rug_pull` to the gateway,
+so every call comes back 503 and the demo denies everything. Catching that here
+costs no ports and no gateway start.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from cmcp_runtime.catalog.loader import definition_digest
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _load(name: str, relative: str):
+    path = ROOT / relative
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.path.insert(0, str(path.parent))
+    try:
+        spec.loader.exec_module(module)
+    finally:
+        sys.path.pop(0)
+    return module
+
+
+def _catalog_entry(demo: str, tool_name: str) -> dict:
+    entries = json.loads((ROOT / demo / "catalog.json").read_text(encoding="utf-8"))
+    for entry in entries:
+        if entry["tool_name"] == tool_name:
+            return entry
+    pytest.fail(f"{demo}/catalog.json has no entry for {tool_name}")
+
+
+def test_model_gateway_catalog_matches_the_advertised_tool():
+    entry = _catalog_entry("demo-10-model-gateway", "model.chat_completion")
+    approved = entry["approved_definition"]
+    tool = _load("demo_model_server", "demo-10-model-gateway/model_server.py").TOOL
+
+    assert definition_digest(
+        approved["description"],
+        approved.get("input_schema"),
+        approved.get("output_schema"),
+    ) == definition_digest(
+        str(tool.get("description", "")),
+        tool.get("inputSchema") or {},
+        tool.get("outputSchema"),
+    ), (
+        "catalog.json and model_server.py describe different tools. The gateway "
+        "reads that as a rug pull and denies every call."
+    )
+
+
+def test_model_gateway_endpoint_arguments_are_all_in_the_schema():
+    """Every argument the endpoint sends must be an approved property.
+
+    This is the specific gap that shipped: the endpoint sent
+    ``contains_identifiers``, the Cedar policy read it, and neither schema
+    declared it.
+    """
+    entry = _catalog_entry("demo-10-model-gateway", "model.chat_completion")
+    properties = set(entry["approved_definition"]["input_schema"]["properties"])
+
+    source = (ROOT / "demo-10-model-gateway" / "endpoint.py").read_text(encoding="utf-8")
+    start = source.index("    args = {")
+    block = source[start:source.index("\n    }", start)]
+    sent = {line.split('"')[1] for line in block.splitlines() if line.strip().startswith('"')}
+
+    assert sent, "could not parse the endpoint's argument block"
+    assert sent <= properties, f"endpoint sends undeclared arguments: {sorted(sent - properties)}"
+
+
+def test_stored_definition_hashes_are_current():
+    """A hand-edited schema with a stale definition_hash fails at catalog load."""
+    from cmcp_runtime.catalog.loader import _compute_definition_hash
+
+    for catalog in sorted(ROOT.glob("demo-*/catalog.json")):
+        for entry in json.loads(catalog.read_text(encoding="utf-8")):
+            if "definition_hash" not in entry:
+                continue
+            computed = _compute_definition_hash(entry["approved_definition"])
+            assert entry["definition_hash"] == computed, (
+                f"{catalog.parent.name}/{entry['tool_name']}: stale definition_hash"
+            )

web-console/run.py

@@ -87,6 +87,28 @@ def _wait_for_port(port: int, what: str, process=None, timeout: float = 60.0) ->
              + _log_tail(what))
 
 
+def _assert_port_free(port: int, what: str) -> None:
+    """Refuse to start if something already owns the port.
+
+    _wait_for_port() returns as soon as *anything* answers, so a gateway left
+    over from a demo run satisfies it instantly while this console's own
+    gateway dies on a bind error in cmcp.log. The console then scores every
+    assessment against the other gateway's policy bundle and still prints
+    plausible allow/deny lines. Same guard the terminal demos already carry.
+    """
+    try:
+        with socket.create_connection(("127.0.0.1", port), timeout=1):
+            pass
+    except OSError:
+        return
+    sys.exit(
+        f"Port {port} is already in use, so {what} cannot start and this console "
+        f"would be scored against whatever is already listening. Stop it first "
+        f"(a cMCP gateway left over from another demo is the usual cause), then "
+        f"re-run."
+    )
+
+
 def _ensure_submodule() -> None:
     if (EXAMPLE / "agent" / "credit_risk_agent.py").exists():
         return
@@ -129,6 +151,9 @@ def _write_gateway_config() -> None:
 
 def main() -> None:
     _ensure_submodule()
+    _assert_port_free(8080, "the EU credit-risk MCP server")
+    _assert_port_free(8443, "the cMCP gateway")
+    _assert_port_free(int(PORT), "the web console")
     _write_gateway_config()
     server_log = open(HERE / "server.log", "w")
     cmcp_log = open(HERE / "cmcp.log", "w")