feat(admission): quota + admission plane — completes placement + quotas + admission

.gitignore

@@ -21,6 +21,9 @@ artifacts/mesh-heartbeats/*.json
 # MCP ops-surface receipts (sealed per tool call at runtime; dir kept via .gitkeep)
 artifacts/mcp-receipts/*.json
 
+# Admission consumption ledger (per-subject usage, runtime state)
+artifacts/admission-usage.json
+
 # Porter control-plane build outputs (compiled locally, not committed)
 artifacts/porter-shim/shim
 artifacts/cloudshell-hardened-pack/culler/culler

capd/compute-plane.mesh.capd.json

@@ -6,6 +6,7 @@
   "description": "A user develops on a low-mem box and the same workload scales out, seamlessly, over whatever the mesh offers: a k8s service, an HPC/SLURM supercomputer, WASM at the edge, a p2p/hyperswarm mesh, volunteer compute (BOINC / Folding@home / open-HEP-style), or an RLC-style blockchain compute market. The substrate is chosen by the placement broker from per-project/per-account policy and live mesh availability, scaling out where it can and where volunteer compute is offered. Governed and fail-closed: sensitive workloads never land on an untrusted (volunteer/p2p/blockchain) backend, every placement is sealed, and with no allowed+available backend it falls back to local or blocks rather than shipping work somewhere the policy forbids. Configured in the portal dashboard.",
   "links": {
     "broker": "tools/compute_plane.py",
+    "admission": "tools/admission.py",
     "executor": "tools/executor.py",
     "cli": "tools/sourceosctl.py",
     "telemetry": "tools/mesh_telemetry.py",

tools/admission.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Quota + admission — the other two thirds of the Control Plane Agent (placement + quotas +
+admission).
+
+`compute_plane.place()` decides WHERE a workload runs. Admission decides WHETHER a subject may run
+it at all *right now*, under per-account / per-project budgets: how many jobs concurrently, how many
+GPUs, how much spend. This is what makes the platform multi-tenant and cost-governed rather than a
+free-for-all scheduler.
+
+Fail-closed: over budget -> denied, and nothing is placed, granted, or dispatched. Admission is
+checked BEFORE the Grant is minted, so an over-quota subject never even receives a capability. On a
+successful dispatch the consumption is charged; when a job finishes it is released (spend is
+cumulative and not released — you don't get money back).
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+DEFAULT_QUOTA = {"max_concurrent": 4, "gpu_max": 2, "cost_budget": 100.0}
+
+
+class AdmissionController:
+    """Per-subject (or per-project) quotas + a live consumption ledger.
+
+    With a `ledger_path`, consumption persists across processes (so a CLI enforces a real running
+    budget, not a fresh one each invocation)."""
+
+    def __init__(self, quotas: dict[str, dict] | None = None, ledger_path=None):
+        self._quotas = quotas or {}
+        self._ledger_path = Path(ledger_path) if ledger_path else None
+        self._usage: dict[str, dict] = self._load()
+
+    def _load(self) -> dict:
+        if self._ledger_path and self._ledger_path.exists():
+            try:
+                return json.loads(self._ledger_path.read_text())
+            except (OSError, json.JSONDecodeError):
+                return {}
+        return {}
+
+    def _save(self) -> None:
+        if self._ledger_path:
+            self._ledger_path.parent.mkdir(parents=True, exist_ok=True)
+            self._ledger_path.write_text(json.dumps(self._usage, sort_keys=True))
+
+    def quota_for(self, key: str) -> dict:
+        return {**DEFAULT_QUOTA, **self._quotas.get(key, {})}
+
+    def usage(self, key: str) -> dict:
+        return self._usage.setdefault(key, {"concurrent": 0, "gpu": 0, "cost": 0.0})
+
+    def admit(self, key: str, workload: dict, *, cost: float = 1.0) -> dict:
+        """Fail-closed admission check. Returns {admitted, reason, quota, usage, exceeded}."""
+        q = self.quota_for(key)
+        u = self.usage(key)
+        gpu = 1 if workload.get("needs_gpu") else 0
+        checks = {
+            "max_concurrent": u["concurrent"] + 1 <= q["max_concurrent"],
+            "gpu_max": u["gpu"] + gpu <= q["gpu_max"],
+            "cost_budget": round(u["cost"] + cost, 6) <= q["cost_budget"],
+        }
+        exceeded = sorted(name for name, ok in checks.items() if not ok)
+        return {"admitted": not exceeded,
+                "reason": "within budget" if not exceeded else f"quota exceeded: {', '.join(exceeded)}",
+                "quota": q, "usage": dict(u), "exceeded": exceeded}
+
+    def charge(self, key: str, workload: dict, *, cost: float = 1.0) -> dict:
+        """Record a dispatch against the subject's budget."""
+        u = self.usage(key)
+        u["concurrent"] += 1
+        u["gpu"] += 1 if workload.get("needs_gpu") else 0
+        u["cost"] = round(u["cost"] + cost, 6)
+        self._save()
+        return dict(u)
+
+    def release(self, key: str, workload: dict) -> dict:
+        """A job finished: free the concurrency + GPU it held. Spend stays spent."""
+        u = self.usage(key)
+        u["concurrent"] = max(0, u["concurrent"] - 1)
+        u["gpu"] = max(0, u["gpu"] - (1 if workload.get("needs_gpu") else 0))
+        self._save()
+        return dict(u)
+
+
+if __name__ == "__main__":
+    import json
+    ac = AdmissionController({"spiffe://sourceos/agent/dev": {"gpu_max": 1, "max_concurrent": 2}})
+    key = "spiffe://sourceos/agent/dev"
+    gpu_wl = {"needs_gpu": True}
+    print(json.dumps(ac.admit(key, gpu_wl), indent=2))
+    ac.charge(key, gpu_wl)
+    # a second GPU job now exceeds gpu_max=1
+    print(json.dumps(ac.admit(key, gpu_wl), indent=2))

tools/executor.py

@@ -146,22 +146,32 @@ def execute(workload: dict, decision: dict, grant: dict, *, session_id: str, ver
 
 
 def run_spine(workload: dict, policy: dict, *, registry, binding: dict, capability: dict,
-              attestation: dict, constraints: dict, signer, verifier, apply: bool = False) -> dict:
-    """The one call that runs a workload governed across the mesh: place -> grant -> verify -> execute.
+              attestation: dict, constraints: dict, signer, verifier, apply: bool = False,
+              admission=None, admission_key: str | None = None, cost: float = 1.0) -> dict:
+    """The one call that runs a workload governed across the mesh: admit -> place -> grant -> verify
+    -> execute -> charge.
 
-    Returns the full trace. If the plane blocks placement (fail-closed), no Grant is issued and
-    nothing is dispatched.
+    Returns the full trace. Fail-closed at every stage: over quota -> denied (no Grant minted); the
+    plane blocks placement -> blocked; neither issues a Grant nor dispatches anything.
     """
     import compute_plane as cp
+    akey = admission_key or binding["spiffe_id"]
+    if admission is not None:
+        adm = admission.admit(akey, workload, cost=cost)
+        if not adm["admitted"]:
+            return {"status": "denied", "admission": adm}
     decision = cp.place(workload, policy, registry.availability())
     if not decision.get("backend"):
         return {"status": "blocked", "decision": decision}
     grant = grant_mod.issue_grant(binding=binding, capability=capability, decision=decision,
                                   attestation=attestation, constraints=constraints, signer=signer)
     execution = execute(workload, decision, grant, session_id=binding["session_id"],
                         verifier=verifier, apply=apply)
-    return {"status": "ran", "backend": decision["backend"], "decision": decision,
-            "grant_id": grant["grant_id"], "execution": execution}
+    result = {"status": "ran", "backend": decision["backend"], "decision": decision,
+              "grant_id": grant["grant_id"], "execution": execution}
+    if admission is not None:
+        result["admission"] = admission.charge(akey, workload, cost=cost)
+    return result
 
 
 if __name__ == "__main__":

tools/sourceosctl.py

@@ -25,13 +25,15 @@
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
+import admission as adm  # noqa: E402
 import compute_plane as cp  # noqa: E402
 import executor as ex  # noqa: E402
 import mcp_a2a_grant as g  # noqa: E402
 import mesh_telemetry as mt  # noqa: E402
 
 ROOT = Path(__file__).resolve().parent.parent
 HEARTBEATS = ROOT / "artifacts" / "mesh-heartbeats"
+LEDGER = ROOT / "artifacts" / "admission-usage.json"
 _DEV_KEY = "dev-only-key-not-for-production"
 
 
@@ -48,7 +50,7 @@ def _sha(s: str) -> str:
 
 
 def run_workload(*, name, command, effect, sensitivity, scalable, gpu, image, subject,
-                 heartbeats_dir, key, apply=False, dry=False) -> dict:
+                 heartbeats_dir, key, apply=False, dry=False, admission=None, cost=1.0) -> dict:
     """Core of `run` — testable without the CLI. Returns the full spine trace (or a placement)."""
     reg = mt.MeshRegistry.from_dir(heartbeats_dir)
     workload = {"name": name, "command": command, "effect": effect, "sensitivity": sensitivity,
@@ -63,17 +65,23 @@ def run_workload(*, name, command, effect, sensitivity, scalable, gpu, image, su
     attestation = g.attestation_bundle(spiffe_id=subject, aum_digest=aum, tpm_valid=True, cosign_valid=True)
     return ex.run_spine(workload, policy, registry=reg, binding=binding, capability=capability,
                         attestation=attestation, constraints={"ops_allow": ["exec.run"]},
-                        signer=g.hmac_signer(key), verifier=g.hmac_verifier(key), apply=apply)
+                        signer=g.hmac_signer(key), verifier=g.hmac_verifier(key), apply=apply,
+                        admission=admission, admission_key=subject, cost=cost)
 
 
 def cmd_run(args) -> int:
     if _dev_mode():
         print("!  DEV MODE: no SOURCEOS_SIGNING_KEY set — synthesizing a dev attestation + HMAC key. "
               "Not for production.", file=sys.stderr)
+    admission = None if args.dry else adm.AdmissionController(ledger_path=LEDGER)
     out = run_workload(name=args.name, command=args.command, effect=args.effect,
                        sensitivity=args.sensitivity, scalable=not args.no_scale, gpu=args.gpu,
                        image=args.image, subject=args.subject, heartbeats_dir=HEARTBEATS,
-                       key=_key(), apply=args.apply, dry=args.dry)
+                       key=_key(), apply=args.apply, dry=args.dry, admission=admission, cost=args.cost)
+    if out["status"] == "denied":
+        a = out["admission"]
+        print(f"DENIED (fail-closed): {a['reason']}  [usage {a['usage']} vs quota {a['quota']}]")
+        return 4
     if out["status"] == "blocked":
         print(f"BLOCKED (fail-closed): {out['decision']['reason']}")
         return 3
@@ -85,6 +93,15 @@ def cmd_run(args) -> int:
     print(f"ran on: {out['backend']}  ({d['backend_trust']})  via grant {out['grant_id']}")
     print(f"dispatch: {e['dispatch']['kind']} (applied={e['dispatch'].get('applied')})")
     print(f"sealed receipt: {e['receipt']['receipt_digest']}")
+    if out.get("admission"):
+        print(f"quota usage: {out['admission']}")
+    return 0
+
+
+def cmd_quota(args) -> int:
+    ac = adm.AdmissionController(ledger_path=LEDGER)
+    print(f"quota for {args.subject}: {ac.quota_for(args.subject)}")
+    print(f"usage:     {ac.usage(args.subject)}")
     return 0
 
 
@@ -125,6 +142,7 @@ def build_parser() -> argparse.ArgumentParser:
     r.add_argument("--gpu", action="store_true", help="workload needs a GPU")
     r.add_argument("--no-scale", action="store_true", help="keep it small (non-scalable)")
     r.add_argument("--subject", default="spiffe://sourceos/agent/dev", help="the requesting subject SPIFFE id")
+    r.add_argument("--cost", type=float, default=1.0, help="cost units to charge against the subject's budget")
     r.add_argument("--apply", action="store_true", help="actually execute (local subprocess / kubectl apply)")
     r.add_argument("--dry", action="store_true", help="only decide placement; dispatch nothing")
     r.set_defaults(func=cmd_run)
@@ -144,6 +162,10 @@ def build_parser() -> argparse.ArgumentParser:
 
     c = sub.add_parser("commons", help="the reproducible knowledge commons")
     c.set_defaults(func=cmd_commons)
+
+    q = sub.add_parser("quota", help="show a subject's quota + current usage")
+    q.add_argument("--subject", default="spiffe://sourceos/agent/dev")
+    q.set_defaults(func=cmd_quota)
     return p
 
 

tools/test_admission.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""Tests for quota + admission. Load-bearing: fail-closed over budget (concurrent / gpu / cost),
+correct charge/release accounting, per-subject isolation, and that the spine actually refuses to
+place/grant/dispatch an over-quota subject."""
+import admission as adm
+import executor as ex
+import mcp_a2a_grant as g
+import mesh_telemetry as mt
+
+KEY = b"adm-test-key"
+DEV = "spiffe://sourceos/agent/dev"
+
+
+def test_admits_within_budget():
+    ac = adm.AdmissionController()
+    r = ac.admit(DEV, {"needs_gpu": True}, cost=1.0)
+    assert r["admitted"] and r["exceeded"] == []
+
+
+def test_denies_over_concurrency():
+    ac = adm.AdmissionController({DEV: {"max_concurrent": 1}})
+    ac.charge(DEV, {})
+    r = ac.admit(DEV, {})
+    assert not r["admitted"] and "max_concurrent" in r["exceeded"]
+
+
+def test_denies_over_gpu():
+    ac = adm.AdmissionController({DEV: {"gpu_max": 1}})
+    ac.charge(DEV, {"needs_gpu": True})
+    r = ac.admit(DEV, {"needs_gpu": True})
+    assert not r["admitted"] and "gpu_max" in r["exceeded"]
+
+
+def test_denies_over_cost_budget():
+    ac = adm.AdmissionController({DEV: {"cost_budget": 5.0}})
+    ac.charge(DEV, {}, cost=4.5)
+    r = ac.admit(DEV, {}, cost=1.0)  # 4.5 + 1.0 = 5.5 > 5.0
+    assert not r["admitted"] and "cost_budget" in r["exceeded"]
+
+
+def test_release_frees_concurrency_and_gpu_but_not_spend():
+    ac = adm.AdmissionController()
+    ac.charge(DEV, {"needs_gpu": True}, cost=3.0)
+    ac.release(DEV, {"needs_gpu": True})
+    u = ac.usage(DEV)
+    assert u["concurrent"] == 0 and u["gpu"] == 0 and u["cost"] == 3.0  # spend stays spent
+
+
+def test_subjects_have_isolated_budgets():
+    ac = adm.AdmissionController({DEV: {"max_concurrent": 1}})
+    ac.charge(DEV, {})
+    other = "spiffe://sourceos/agent/other"
+    assert ac.admit(DEV, {})["admitted"] is False       # dev is full
+    assert ac.admit(other, {})["admitted"] is True       # other is independent
+
+
+def test_default_quota_applies_to_unknown_subject():
+    ac = adm.AdmissionController()
+    assert ac.quota_for("spiffe://who/dis") == adm.DEFAULT_QUOTA
+
+
+def test_ledger_persists_consumption_across_controllers():
+    import pathlib
+    import tempfile
+    with tempfile.TemporaryDirectory() as td:
+        p = pathlib.Path(td) / "usage.json"
+        ac1 = adm.AdmissionController({DEV: {"max_concurrent": 1}}, ledger_path=p)
+        ac1.charge(DEV, {})
+        ac2 = adm.AdmissionController({DEV: {"max_concurrent": 1}}, ledger_path=p)  # fresh process
+        assert ac2.admit(DEV, {})["admitted"] is False  # sees ac1's charge on disk
+
+
+# ── integration with the spine ───────────────────────────────────────────────────────
+def _spine_args(gpu):
+    aum = "sha256:" + "ab" * 32
+    reg = mt.MeshRegistry()
+    reg.heartbeat("slurm", "hpc-slurm", 100)
+    return dict(
+        workload={"name": "t", "sensitivity": "normal", "scalable": True, "needs_gpu": gpu,
+                  "effect": "compute", "command": "echo hi"},
+        policy={}, registry=reg,
+        binding={"spiffe_id": DEV, "aum_digest": aum, "session_id": "sess_adm01"},
+        capability={"kind": "mcp_tool", "capability_ref": "capd://caps.x",
+                    "capability_digest": "sha256:" + "cd" * 32, "effect": "compute"},
+        attestation=g.attestation_bundle(spiffe_id=DEV, aum_digest=aum, tpm_valid=True, cosign_valid=True),
+        constraints={"ops_allow": ["exec.run"]},
+        signer=g.hmac_signer(KEY), verifier=g.hmac_verifier(KEY))
+
+
+def test_spine_denies_over_quota_before_granting():
+    ac = adm.AdmissionController({DEV: {"gpu_max": 1}})
+    args = _spine_args(gpu=True)
+    first = ex.run_spine(**args, admission=ac, cost=1.0)
+    assert first["status"] == "ran" and first["admission"]["gpu"] == 1
+    second = ex.run_spine(**_spine_args(gpu=True), admission=ac, cost=1.0)
+    assert second["status"] == "denied"          # over gpu_max
+    assert "grant_id" not in second               # no Grant minted, nothing dispatched
+    assert "gpu_max" in second["admission"]["exceeded"]
+
+
+if __name__ == "__main__":
+    import sys
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    for fn in fns:
+        fn()
+    print(f"ok: {len(fns)} admission tests passed")
+    sys.exit(0)

tools/validate.py

@@ -31,6 +31,7 @@
     "tools/mcp_ops_server.py",
     "tools/executor.py",
     "tools/sourceosctl.py",
+    "tools/admission.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.