feat(executor): execution spine — place → grant → verify → dispatch → receipt + sourceosctl CLI

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 mesh-demo grant commons mcp
+.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run
 
 validate: ## repo hygiene + CapD validity
 	python3 tools/validate.py
@@ -27,6 +27,12 @@ grant: ## demo the zero-trust attach flow: Attest -> Decide -> Grant -> verify-a
 commons: ## reproducible knowledge commons: ingest the estate's CapDs + workloads as citable records
 	python3 tools/commons.py
 
+spine: ## run the full execution spine demo: place -> grant -> verify -> dispatch -> sealed receipt
+	cd tools && python3 executor.py
+
+run: ## sourceosctl: run a workload governed across the mesh (e.g. make run ARGS="run --gpu --command 'python train.py'")
+	python3 tools/sourceosctl.py $(ARGS)
+
 onboard: ## bring up a workstation: local sovereign forge + local cluster + sourceosctl
 	@echo "[continuum] onboard — scaffold: wires Gitea bring-up + kind/k3s + sourceos-devtools/sourceosctl"
 

capd/compute-plane.mesh.capd.json

@@ -6,6 +6,10 @@
   "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",
+    "executor": "tools/executor.py",
+    "cli": "tools/sourceosctl.py",
+    "telemetry": "tools/mesh_telemetry.py",
+    "grant_authority": "tools/mcp_a2a_grant.py",
     "portal": "tools/portal_server.py",
     "agent_surface": "tools/mcp_ops_server.py",
     "reference_pattern": "BOINC computing-preferences + IBM Parallel Environment (HPC toolkit) + Docker volume drivers — met and bettered: one governed plane across all substrates, not per-substrate config silos"

docs/EXECUTION_SPINE.md

@@ -0,0 +1,54 @@
+# The execution spine — from a low-mem box to anywhere, governed
+
+This is what makes the compute plane a *platform* and not a planner. One command from a low-mem box
+places a workload, mints and verifies a zero-trust Grant, and **actually dispatches** it — to
+whatever substrate the mesh offers — sealing a receipt. Fail-closed at every step.
+
+```
+sourceosctl run --command "python train.py" --gpu --sensitivity sensitive
+```
+```
+ran on: hpc-slurm  (trusted)  via grant grant_caed7e42fff04099
+dispatch: hpc-slurm (applied=False)
+sealed receipt: sha256:3d504bec…
+```
+
+## The flow (`tools/executor.py` · `tools/sourceosctl.py`)
+
+```
+mesh_telemetry           compute_plane        mcp_a2a_grant           mcp_a2a_grant        executor
+  (live avail.)   ──►   place() [Decide]  ──►  issue_grant() [Grant] ─► verify_grant() ─►  dispatch ─► sealed receipt
+   heartbeats            per-policy,            attest→decide→grant      [Policy Gate]      per-backend adapter
+   TTL fail-closed       per-availability       session-bound, signed    fail-closed         + receipt
+```
+
+1. **Telemetry** — the live mesh (`sourceosctl mesh`); stale nodes count zero.
+2. **Decide** — `place()` picks a backend under per-project/per-account policy. A sensitive workload
+   never lands on an untrusted (volunteer/p2p/blockchain) backend; if nothing compliant is live it
+   **blocks**, it doesn't degrade.
+3. **Grant** — `issue_grant()` attests, then mints a canonical session-bound Grant (see
+   `docs/CLOUDSHELL_FOG.md` and the vendored `schemas/a2a/`).
+4. **Gate** — `verify_grant()` re-checks the Grant at the node: signature, session, expiry,
+   attestation, effect. No valid Grant → **`DispatchRefused`**, nothing runs.
+5. **Dispatch** — a per-backend adapter runs it:
+   - **local** — a real subprocess (`--apply`).
+   - **k8s** — a real, Grant-labelled `batch/v1` Job manifest (applied via `kubectl` when a cluster
+     is reachable, else emitted).
+   - **hpc-slurm · wasm-edge · p2p-mesh · volunteer-boinc · blockchain-rlc** — the substrate-specific
+     descriptor to hand that scheduler over a Grant-bound channel.
+6. **Receipt** — every dispatch is hash-sealed.
+
+## CLI
+
+| command | does |
+|---|---|
+| `sourceosctl mesh` | what's live in the mesh right now |
+| `sourceosctl place --gpu --sensitivity sensitive` | where would this land? (dry, no dispatch) |
+| `sourceosctl run … [--apply]` | place → grant → verify → dispatch; `--apply` actually executes |
+| `sourceosctl commons` | the reproducible knowledge commons |
+
+**Production vs dev.** In production the Grant is signed by the Key Authority (HSM/KMS) and the
+`AttestationBundle` comes from the node's TPM/TEE + cosign. Without `SOURCEOS_SIGNING_KEY` set, the
+CLI runs in **DEV MODE** — it synthesizes a dev attestation + HMAC key and says so loudly on stderr.
+
+See `tools/test_executor.py` (adapters + fail-closed + full spine) and `tools/test_sourceosctl.py`.

tools/executor.py

@@ -0,0 +1,189 @@
+#!/usr/bin/env python3
+"""The execution spine — turns a governed placement decision + a verified Grant into an ACTUAL
+dispatch. This is what makes the compute plane a *platform* and not a planner.
+
+The whole flow, closed:
+
+    heartbeat -> place() [Decide] -> issue_grant() [Grant] -> verify_grant() [Gate] -> DISPATCH
+    -> sealed execution receipt
+
+Dispatch is pluggable per substrate: run it locally as a subprocess, emit a real k8s Job manifest
+(and apply it when a cluster is reachable), or emit the substrate-specific descriptor you'd hand to
+an HPC/SLURM queue, a WASM edge, a p2p/hyperswarm peer, a volunteer grid, or a blockchain compute
+market. Every dispatch is **fail-closed on the Grant**: the fog-node Policy Gate re-verifies the
+Grant (signature, session binding, expiry, attestation, effect) before anything runs — no valid
+Grant, no dispatch — and every dispatch is sealed into a tamper-evident receipt.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import shlex
+import shutil
+import subprocess
+from datetime import datetime, timezone
+
+import mcp_a2a_grant as grant_mod
+
+
+def _seal(body: dict) -> str:
+    return "sha256:" + hashlib.sha256(
+        json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+class DispatchRefused(Exception):
+    """The Grant did not verify at the node — nothing is dispatched (fail-closed)."""
+
+
+# ── backend adapters ─────────────────────────────────────────────────────────────────
+class LocalAdapter:
+    """Runs the workload as a local subprocess (the dev's own box)."""
+    backend = "local"
+
+    def dispatch(self, workload, decision, grant, *, apply):
+        cmd = workload.get("command")
+        if not cmd:
+            return {"kind": "local", "applied": False, "reason": "no command to run"}
+        if not apply:
+            return {"kind": "local", "applied": False, "planned": cmd}
+        proc = subprocess.run(shlex.split(cmd), capture_output=True, text=True, timeout=120)
+        return {"kind": "local", "applied": True, "exit_code": proc.returncode,
+                "stdout": proc.stdout[-2000:], "stderr": proc.stderr[-2000:]}
+
+
+class K8sAdapter:
+    """Emits a real batch/v1 Job manifest, Grant-labelled; applies it when a cluster is reachable."""
+    backend = "k8s"
+
+    def manifest(self, workload, decision, grant):
+        res = workload.get("resource", {})
+        requests = {}
+        if res.get("cpu"):
+            requests["cpu"] = str(res["cpu"])
+        if res.get("mem"):
+            requests["memory"] = str(res["mem"])
+        limits = dict(requests)
+        if workload.get("needs_gpu"):
+            limits["nvidia.com/gpu"] = "1"
+        container = {"name": "workload",
+                     "image": workload.get("image", "busybox:1.36"),
+                     "command": shlex.split(workload["command"]) if workload.get("command") else ["true"]}
+        if requests or limits:
+            container["resources"] = {k: v for k, v in (("requests", requests), ("limits", limits)) if v}
+        return {
+            "apiVersion": "batch/v1", "kind": "Job",
+            "metadata": {"generateName": f"{workload.get('name', 'wl')}-",
+                         "labels": {"sourceos.io/grant-id": grant["grant_id"],
+                                    "sourceos.io/session": grant["binding"].get("session_id", ""),
+                                    "sourceos.io/backend": "k8s"}},
+            "spec": {"backoffLimit": 0, "ttlSecondsAfterFinished": 3600,
+                     "template": {"metadata": {"labels": {"sourceos.io/grant-id": grant["grant_id"]}},
+                                  "spec": {"restartPolicy": "Never", "containers": [container]}}},
+        }
+
+    def dispatch(self, workload, decision, grant, *, apply):
+        manifest = self.manifest(workload, decision, grant)
+        if apply and shutil.which("kubectl"):
+            proc = subprocess.run(["kubectl", "apply", "-f", "-"], input=json.dumps(manifest),
+                                  capture_output=True, text=True, timeout=60)
+            return {"kind": "k8s", "applied": proc.returncode == 0, "manifest": manifest,
+                    "kubectl": proc.stdout.strip() or proc.stderr.strip()}
+        return {"kind": "k8s", "applied": False, "manifest": manifest}
+
+
+class DescriptorAdapter:
+    """For substrates dispatched by handing a descriptor to their own scheduler/mesh: HPC/SLURM,
+    WASM edge, p2p/hyperswarm, volunteer grid, blockchain compute market."""
+    def __init__(self, backend: str):
+        self.backend = backend
+
+    def dispatch(self, workload, decision, grant, *, apply):
+        return {"kind": self.backend, "applied": False,
+                "descriptor": {"backend": self.backend, "executor_ref": grant["capability"].get("executor_ref"),
+                               "command": workload.get("command"), "image": workload.get("image"),
+                               "resource": workload.get("resource", {}), "needs_gpu": workload.get("needs_gpu", False),
+                               "grant_id": grant["grant_id"], "session": grant["binding"].get("session_id")},
+                "note": f"hand this descriptor to the {self.backend} scheduler over a Grant-bound channel"}
+
+
+def default_adapters() -> dict:
+    ad = {LocalAdapter().backend: LocalAdapter(), K8sAdapter().backend: K8sAdapter()}
+    for b in ("hpc-slurm", "wasm-edge", "p2p-mesh", "volunteer-boinc", "blockchain-rlc"):
+        ad[b] = DescriptorAdapter(b)
+    return ad
+
+
+def execute(workload: dict, decision: dict, grant: dict, *, session_id: str, verifier,
+            adapters: dict | None = None, apply: bool = False) -> dict:
+    """Enforce the Grant at the node, then dispatch to the decided backend, then seal a receipt.
+
+    Fail-closed: if verify_grant denies (bad sig / wrong session / expired / attestation / effect),
+    raise DispatchRefused — nothing runs.
+    """
+    adapters = adapters or default_adapters()
+    check = grant_mod.verify_grant(grant, session_id=session_id, verifier=verifier,
+                                   requested_effect=workload.get("effect"))
+    if not check["result"]["valid"]:
+        raise DispatchRefused(check["result"]["reason"])
+
+    backend = decision.get("backend")
+    adapter = adapters.get(backend)
+    if adapter is None:
+        raise DispatchRefused(f"no execution adapter for backend {backend!r}")
+
+    dispatch = adapter.dispatch(workload, decision, grant, apply=apply)
+    receipt = {"spine": "sourceos-continuum.executor.v1", "backend": backend,
+               "grant_id": grant["grant_id"], "session": session_id,
+               "applied": dispatch.get("applied", False), "dispatch_kind": dispatch.get("kind"),
+               "grant_check": check["check_id"], "at": _now_iso()}
+    receipt["receipt_digest"] = _seal(receipt)
+    return {"status": "dispatched", "backend": backend, "dispatch": dispatch,
+            "grant_check": check, "receipt": receipt}
+
+
+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.
+
+    Returns the full trace. If the plane blocks placement (fail-closed), no Grant is issued and
+    nothing is dispatched.
+    """
+    import compute_plane as cp
+    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}
+
+
+if __name__ == "__main__":
+    # demo: run the full spine for a sensitive GPU workload against a live mesh (dry-run dispatch).
+    import mesh_telemetry as mt
+    key = b"demo-executor-key"
+    reg = mt.MeshRegistry()
+    reg.heartbeat("slurm-1", "hpc-slurm", 100)
+    reg.heartbeat("boinc-1", "volunteer-boinc", 500)
+    att = grant_mod.attestation_bundle(spiffe_id="spiffe://sourceos/agent/demo",
+                                       aum_digest="sha256:" + "ab" * 32, tpm_valid=True, cosign_valid=True)
+    out = run_spine(
+        {"name": "train", "sensitivity": "sensitive", "scalable": True, "needs_gpu": True,
+         "effect": "compute", "image": "ghcr.io/sourceos/trainer:1", "command": "python train.py"},
+        {"require_attestation": True}, registry=reg,
+        binding={"spiffe_id": "spiffe://sourceos/agent/demo", "aum_digest": "sha256:" + "ab" * 32,
+                 "session_id": "sess_demo1"},
+        capability={"kind": "mcp_tool", "capability_ref": "capd://caps.compute.mesh-plane",
+                    "capability_digest": "sha256:" + "cd" * 32, "effect": "compute"},
+        attestation=att, constraints={"ops_allow": ["exec.run"]},
+        signer=grant_mod.hmac_signer(key), verifier=grant_mod.hmac_verifier(key))
+    print(json.dumps({"status": out["status"], "backend": out["backend"],
+                      "grant_id": out["grant_id"],
+                      "dispatch": out["execution"]["dispatch"]["kind"],
+                      "receipt": out["execution"]["receipt"]["receipt_digest"]}, indent=2))