feat(mcp): governed MCP ops surface — agent-native, fail-closed, hash-sealed (march #1)

.gitignore

@@ -18,6 +18,9 @@ artifacts/gate-decisions/*.json
 # Live mesh telemetry heartbeats (emitted by nodes at runtime, not committed; dir kept via .gitkeep)
 artifacts/mesh-heartbeats/*.json
 
+# MCP ops-surface receipts (sealed per tool call at runtime; dir kept via .gitkeep)
+artifacts/mcp-receipts/*.json
+
 # Porter control-plane build outputs (compiled locally, not committed)
 artifacts/porter-shim/shim
 artifacts/cloudshell-hardened-pack/culler/culler

Makefile

@@ -1,6 +1,6 @@
 # SourceOS Continuum — lifecycle entry points.
 # Control-plane targets delegate to Makefile.porter (the rehomed Porter control plane).
-.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute
+.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp
 
 validate: ## repo hygiene + CapD validity
 	python3 tools/validate.py
@@ -48,3 +48,6 @@ promotion-gate: ## rollout gate: require an APPROVE review verdict (fail-closed,
 
 rollout: promotion-gate ## promote local → scale-up cluster (hyperswarm), gated on an APPROVE review verdict
 	@echo "[continuum] rollout — promote via caps.infra.cluster-scaleup.hyperswarm (promotion gate passed)"
+
+mcp: ## run the governed MCP ops surface (stdio; add ADD an agent client, e.g. Claude Code). Guarded tools fail-closed unless CONTINUUM_MCP_ALLOW_GUARDED=1
+	python3 tools/mcp_ops_server.py

artifacts/mcp-receipts/.gitkeep

@@ -0,0 +1 @@
+# Sealed per-tool-call receipts from the MCP ops surface (tools/mcp_ops_server.py) land here at runtime.

tools/mcp_ops_server.py

@@ -0,0 +1,233 @@
+#!/usr/bin/env python3
+"""Governed MCP ops surface for sourceos-continuum — agent-native, fail-closed, hash-sealed.
+
+A minimal stdio JSON-RPC (MCP) server, no external dependencies, exposing continuum's OWNED
+control-plane surface as agent-callable tools so an agent (Claude Code, Cursor, ...) can drive
+the platform conversationally. This is superiority-march move #1: it closes the #1 competitive
+gap (agent-native MCP ops surface) while keeping the properties the SaaS incumbents lack.
+
+What makes this different from Qovery/Port/Render MCP surfaces (all SaaS, audit-logged):
+ - FAIL-CLOSED by default. Read tools run. A `guarded` (mutating/decisioning) tool is REFUSED
+ unless an explicit policy grant is present (CONTINUUM_MCP_ALLOW_GUARDED) — it never
+ executes-then-audits. Refusal is the default, not the exception.
+ - A HASH-SEALED receipt on EVERY tool call — allow or refuse — emitted to the evidence
+ bundle. A tamper-evident ledger, not a mutable audit log.
+ - FULLY OPEN: MIT, self-hosted, stdio transport, zero external deps, scale-to-zero (spawned
+ on demand, exits when the client disconnects).
+
+Boundary (CONTINUUM_SCOPE.md): this orchestrates + governs continuum's owned surface (CapD
+capabilities, lifecycle, evidence, the promotion gate). It consumes — never reimplements — the
+review verdict (prophet-platform) and the source/registry (Gitea/zot).
+"""
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parent.parent
+PROTOCOL_VERSION = "2025-06-18"
+SERVER = {"name": "sourceos-continuum-ops", "version": "0.1.0"}
+EVIDENCE_DIR = _ROOT / "artifacts" / "mcp-receipts"
+GUARDED_GRANT_ENV = "CONTINUUM_MCP_ALLOW_GUARDED"
+
+
+def _load(mod_name: str, rel: str):
+ spec = importlib.util.spec_from_file_location(mod_name, _ROOT / rel)
+ mod = importlib.util.module_from_spec(spec)
+ sys.modules[mod_name] = mod
+ spec.loader.exec_module(mod)
+ return mod
+
+
+def _seal(body: dict) -> str:
+ return "sha256:" + hashlib.sha256(
+ json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
+
+
+def _digest(obj) -> str:
+ return "sha256:" + hashlib.sha256(
+ json.dumps(obj, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")).hexdigest()
+
+
+def _emit_receipt(tool: str, arguments: dict, mode: str, decision: str, result) -> dict:
+ """Seal and persist a receipt for one tool call. Every call is recorded — this is the
+ ledger that makes the surface auditable-by-construction, not by opt-in logging."""
+ receipt = {
+ "surface": "sourceos-continuum.mcp_ops.v1",
+ "tool": tool,
+ "mode": mode,
+ "decision": decision, # "allow" | "refuse"
+ "arguments_digest": _digest(arguments),
+ "result_digest": _digest(result),
+ "decided_at": datetime.now(timezone.utc).isoformat(),
+ }
+ receipt["receipt_digest"] = _seal(receipt)
+ try:
+ EVIDENCE_DIR.mkdir(parents=True, exist_ok=True)
+ stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S_%f")
+ (EVIDENCE_DIR / f"{tool}.{decision}.{stamp}.json").write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n")
+ except OSError:
+ pass # a read-only estate must not break the call; the receipt is still returned inline
+ return receipt
+
+
+# ── tool handlers (pure reads over continuum's owned surface) ─────────────────────────
+
+def _tool_list_capabilities(_args: dict) -> dict:
+ caps = []
+ capd_dir = _ROOT / "capd"
+ for f in sorted(capd_dir.glob("*.capd.json")) if capd_dir.is_dir() else []:
+ try:
+ data = json.loads(f.read_text())
+ caps.append({"file": f.name, "capability_id": data.get("capability_id"),
+ "kind": data.get("kind"), "status": data.get("status"),
+ "composes_with": data.get("composes_with", {})})
+ except (OSError, json.JSONDecodeError):
+ continue
+ return {"capabilities": caps}
+
+
+def _tool_lifecycle_status(_args: dict) -> dict:
+ return {"lifecycle": [
+ {"stage": "onboard", "owned": True},
+ {"stage": "develop", "owned": True},
+ {"stage": "cloud-native-test", "owned": True},
+ {"stage": "rollout", "owned": True, "gate": "promotion_gate (fail-closed on APPROVE verdict)"},
+ ], "source": "docs/LIFECYCLE.md"}
+
+
+def _tool_list_evidence(args: dict) -> dict:
+ limit = int(args.get("limit", 20))
+ bundles = []
+ for d in ("gate-decisions", "mcp-receipts"):
+ p = _ROOT / "artifacts" / d
+ if p.is_dir():
+ for f in sorted(p.glob("*.json"), reverse=True)[:limit]:
+ bundles.append({"bundle": d, "name": f.name})
+ return {"evidence": bundles[:limit]}
+
+
+def _tool_promotion_gate(args: dict) -> dict:
+ """Guarded: run the promotion gate on a consumed review verdict. Emits a gate decision, so
+ it is policy-gated. Reuses tools/promotion_gate.py (single source; not reimplemented)."""
+ verdict = args.get("verdict")
+ if not isinstance(verdict, dict):
+ return {"error": "verdict (a sealed review receipt object) is required"}
+ pg = _load("promotion_gate", "tools/promotion_gate.py")
+ promote, decision = pg.gate(verdict, _ROOT / "artifacts" / "gate-decisions")
+ return {"promotion": decision["promotion"], "decision": decision}
+
+
+TOOLS = {
+ "continuum_list_capabilities": {
+ "mode": "read",
+ "description": "List the platform's CapD capability contracts (id, kind, status, composition).",
+ "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
+ "handler": _tool_list_capabilities,
+ },
+ "continuum_lifecycle_status": {
+ "mode": "read",
+ "description": "The continuum lifecycle stages (onboard -> develop -> cloud-native-test -> rollout) and their gates.",
+ "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
+ "handler": _tool_lifecycle_status,
+ },
+ "continuum_list_evidence": {
+ "mode": "read",
+ "description": "List the most recent sealed evidence bundles (gate decisions + MCP receipts).",
+ "inputSchema": {"type": "object", "properties": {"limit": {"type": "integer"}}, "additionalProperties": False},
+ "handler": _tool_list_evidence,
+ },
+ "continuum_promotion_gate": {
+ "mode": "guarded",
+ "description": "Run the fail-closed rollout promotion gate over a consumed review verdict. GUARDED: refused unless policy-granted.",
+ "inputSchema": {"type": "object", "properties": {"verdict": {"type": "object"}},
+ "required": ["verdict"], "additionalProperties": False},
+ "handler": _tool_promotion_gate,
+ },
+}
+
+
+def _guarded_allowed() -> bool:
+ return os.environ.get(GUARDED_GRANT_ENV, "") not in ("", "0", "false", "no")
+
+
+def call_tool(name: str, arguments: dict) -> dict:
+ """Dispatch one tool call through the governance gate, sealing a receipt either way.
+ Returns an MCP tools/call result ({content, isError})."""
+ arguments = arguments or {}
+ tool = TOOLS.get(name)
+ if tool is None:
+ result = {"error": f"unknown tool {name!r}"}
+ _emit_receipt(name, arguments, "unknown", "refuse", result)
+ return {"content": [{"type": "text", "text": json.dumps(result)}], "isError": True}
+
+ mode = tool["mode"]
+ if mode == "guarded" and not _guarded_allowed():
+ result = {"refused": True,
+ "reason": f"guarded tool {name!r} is fail-closed; set {GUARDED_GRANT_ENV}=1 "
+ f"(an explicit, audited policy grant) to permit it"}
+ _emit_receipt(name, arguments, mode, "refuse", result)
+ return {"content": [{"type": "text", "text": json.dumps(result)}], "isError": True}
+
+ try:
+ result = tool["handler"](arguments)
+ except Exception as exc: # a tool fault must not crash the server
+ result = {"error": f"{type(exc).__name__}: {exc}"}
+ _emit_receipt(name, arguments, mode, "error", result)
+ return {"content": [{"type": "text", "text": json.dumps(result)}], "isError": True}
+
+ receipt = _emit_receipt(name, arguments, mode, "allow", result)
+ result["_receipt"] = receipt["receipt_digest"] # the sealed evidence for this call
+ return {"content": [{"type": "text", "text": json.dumps(result, indent=2, sort_keys=True)}]}
+
+
+def handle(request: dict) -> dict | None:
+ """Handle one JSON-RPC request. Returns a response dict, or None for notifications."""
+ method = request.get("method")
+ rid = request.get("id")
+ if method == "initialize":
+ return {"jsonrpc": "2.0", "id": rid, "result": {
+ "protocolVersion": PROTOCOL_VERSION,
+ "capabilities": {"tools": {}},
+ "serverInfo": SERVER,
+ "instructions": "Governed continuum ops surface. Read tools run; guarded tools are fail-closed. Every call is sealed.",
+ }}
+ if method in ("notifications/initialized", "notifications/cancelled"):
+ return None
+ if method == "ping":
+ return {"jsonrpc": "2.0", "id": rid, "result": {}}
+ if method == "tools/list":
+ return {"jsonrpc": "2.0", "id": rid, "result": {"tools": [
+ {"name": n, "description": t["description"], "inputSchema": t["inputSchema"],
+ "annotations": {"readOnlyHint": t["mode"] == "read"}}
+ for n, t in TOOLS.items()]}}
+ if method == "tools/call":
+ params = request.get("params") or {}
+ return {"jsonrpc": "2.0", "id": rid, "result": call_tool(params.get("name"), params.get("arguments"))}
+ return {"jsonrpc": "2.0", "id": rid,
+ "error": {"code": -32601, "message": f"method not found: {method}"}}
+
+
+def main() -> int:
+ for line in sys.stdin:
+ line = line.strip()
+ if not line:
+ continue
+ try:
+ request = json.loads(line)
+ except json.JSONDecodeError:
+ continue
+ response = handle(request)
+ if response is not None:
+ sys.stdout.write(json.dumps(response) + "\n")
+ sys.stdout.flush()
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())

tools/test_mcp_ops_server.py

@@ -0,0 +1,105 @@
+"""Coverage for tools/mcp_ops_server.py — the governed MCP ops surface.
+
+Drives the real MCP JSON-RPC methods (initialize / tools/list / tools/call) and proves the
+governance model: read tools run and seal a receipt; a guarded tool is refused fail-closed
+without a policy grant and runs with one; every call — allow or refuse — lands a tamper-evident
+receipt in the evidence bundle.
+"""
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load(name: str, rel: str):
+ spec = importlib.util.spec_from_file_location(name, ROOT / rel)
+ mod = importlib.util.module_from_spec(spec)
+ sys.modules[name] = mod
+ spec.loader.exec_module(mod)
+ return mod
+
+
+srv = _load("mcp_ops_server", "tools/mcp_ops_server.py")
+
+
+@pytest.fixture(autouse=True)
+def _isolate(tmp_path, monkeypatch):
+ monkeypatch.setattr(srv, "EVIDENCE_DIR", tmp_path / "mcp-receipts")
+ monkeypatch.delenv(srv.GUARDED_GRANT_ENV, raising=False)
+
+
+def _call(name, arguments=None):
+ return srv.handle({"jsonrpc": "2.0", "id": 1, "method": "tools/call",
+ "params": {"name": name, "arguments": arguments or {}}})
+
+
+def test_initialize_advertises_the_server():
+ r = srv.handle({"jsonrpc": "2.0", "id": 1, "method": "initialize"})
+ assert r["result"]["protocolVersion"] == srv.PROTOCOL_VERSION
+ assert r["result"]["serverInfo"]["name"] == "sourceos-continuum-ops"
+ assert "tools" in r["result"]["capabilities"]
+
+
+def test_notifications_get_no_response():
+ assert srv.handle({"jsonrpc": "2.0", "method": "notifications/initialized"}) is None
+
+
+def test_tools_list_flags_read_vs_guarded():
+ r = srv.handle({"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
+ by = {t["name"]: t for t in r["result"]["tools"]}
+ assert set(by) == {"continuum_list_capabilities", "continuum_lifecycle_status",
+ "continuum_list_evidence", "continuum_promotion_gate"}
+ assert by["continuum_list_capabilities"]["annotations"]["readOnlyHint"] is True
+ assert by["continuum_promotion_gate"]["annotations"]["readOnlyHint"] is False
+
+
+def test_read_tool_runs_and_seals_a_receipt():
+ r = _call("continuum_list_capabilities")
+ assert "isError" not in r["result"]
+ payload = json.loads(r["result"]["content"][0]["text"])
+ assert "capabilities" in payload and payload["_receipt"].startswith("sha256:")
+ files = list(srv.EVIDENCE_DIR.glob("*.json"))
+ assert len(files) == 1
+ rec = json.loads(files[0].read_text())
+ assert rec["decision"] == "allow"
+ body = {k: v for k, v in rec.items() if k != "receipt_digest"}
+ assert srv._seal(body) == rec["receipt_digest"] # tamper-evident
+
+
+def test_guarded_tool_is_fail_closed_without_a_grant():
+ r = _call("continuum_promotion_gate", {"verdict": {}})
+ assert r["result"]["isError"] is True
+ payload = json.loads(r["result"]["content"][0]["text"])
+ assert payload["refused"] is True and "fail-closed" in payload["reason"]
+ # the refusal is itself sealed
+ refusals = list(srv.EVIDENCE_DIR.glob("*refuse*.json"))
+ assert len(refusals) == 1 and json.loads(refusals[0].read_text())["decision"] == "refuse"
+
+
+def test_guarded_tool_runs_with_an_explicit_grant(monkeypatch):
+ monkeypatch.setenv(srv.GUARDED_GRANT_ENV, "1")
+ pg = _load("promotion_gate", "tools/promotion_gate.py")
+ verdict = {"tool": "prophet-platform.review_gate.v1",
+ "reviewed": {"idempotency_key": "engine@0.4.45"}, "verdict": "APPROVE", "checks": []}
+ verdict["review_digest"] = pg._recompute_seal(verdict)
+ r = _call("continuum_promotion_gate", {"verdict": verdict})
+ assert "isError" not in r["result"], r["result"]
+ payload = json.loads(r["result"]["content"][0]["text"])
+ assert payload["promotion"] == "allow"
+
+
+def test_unknown_tool_is_refused_and_sealed():
+ r = _call("does_not_exist")
+ assert r["result"]["isError"] is True
+ assert list(srv.EVIDENCE_DIR.glob("*.json")) # even the unknown-tool refusal is recorded
+
+
+def test_unknown_method_returns_jsonrpc_error():
+ r = srv.handle({"jsonrpc": "2.0", "id": 1, "method": "bogus/method"})
+ assert r["error"]["code"] == -32601

tools/validate.py

@@ -28,6 +28,7 @@
  "tools/mesh_telemetry.py",
  "tools/mcp_a2a_grant.py",
  "tools/commons.py",
+ "tools/mcp_ops_server.py",
 ]
 CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy")
 # Every CapD in capd/ must carry the core keys and parse — not just the flagship control-plane one.