feat(control-loop): governed autonomous self-healing — detect AND heal

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 spine run
+.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run loop
 
 validate: ## repo hygiene + CapD validity
 	python3 tools/validate.py
@@ -33,6 +33,9 @@ spine: ## run the full execution spine demo: place -> grant -> verify -> dispatc
 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)
 
+loop: ## autonomous control loop demo: sense -> governed spine -> act, once per cooldown
+	cd tools && python3 control_loop.py
+
 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/self-healing-loop.mesh.capd.json

@@ -0,0 +1,30 @@
+{
+  "capability_id": "caps.compute.self-healing-loop@0.1.0",
+  "kind": "compute.autonomous-control-loop",
+  "status": "experimental",
+  "name": "Governed autonomous control loop — detect AND heal",
+  "description": "Closes the loop from detection to remediation, under governance. It senses conditions and, for each one it may act on, runs the SAME governed spine a developer runs (admission -> place -> grant -> verify -> execute), sealing every action. Safe to run unattended: it never acts without a valid Grant or over quota; it decides a persistent condition ONCE per cooldown (no remediation storms); and one bad condition (block, deny, or raise) is recorded and skipped rather than killing the loop. This is the difference between a platform that pages a human and one that self-heals.",
+  "links": {
+    "engine": "tools/control_loop.py",
+    "spine": "tools/executor.py",
+    "placement": "tools/compute_plane.py",
+    "admission": "tools/admission.py",
+    "grant_authority": "tools/mcp_a2a_grant.py",
+    "ledger": "artifacts/gate-decisions + mcp-receipts (sealed)",
+    "reference_pattern": "escalation-suppression / decide-a-standing-condition-once-per-cooldown; detect != control-loop"
+  },
+  "composes_with": {
+    "compute_plane": "caps.compute.mesh-plane@0.1.0",
+    "fog": "caps.compute.cloudshell-fog@0.1.0",
+    "control_plane": "caps.infra.paas.continuum-local@0.1.0",
+    "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0"
+  },
+  "policy": {
+    "governed_remediation": true,
+    "never_acts_ungoverned": true,
+    "decide_once_per_cooldown": true,
+    "fail_closed": true,
+    "evidence_emitting": true,
+    "one_bad_condition_never_kills_the_loop": true
+  }
+}

tools/control_loop.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Governed autonomous control loop — detect ≠ heal, until now.
+
+Most platforms detect a problem and page a human. This closes the loop: it senses conditions, and
+for each one it may act on, it runs the SAME governed spine a developer runs
+(admission → place → grant → verify → execute), sealing every action.
+
+Three properties make it safe enough to run unattended:
+
+  * It never acts ungoverned — every remediation goes through the grant + quota path. No valid
+    Grant, or over quota, means nothing runs; the loop records that and carries on.
+  * It decides a persistent condition ONCE per cooldown — no remediation storms (the same
+    escalation-suppression discipline the estate uses elsewhere: decide a standing condition once,
+    not every tick).
+  * One bad condition never kills the loop — a remediation that blocks, is denied, or even raises is
+    caught, recorded, and the loop moves to the next condition.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from datetime import datetime, timezone
+
+import executor as ex
+
+
+def _seal(body: dict) -> str:
+    return "sha256:" + hashlib.sha256(
+        json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
+
+
+def _iso(ts: float) -> str:
+    return datetime.fromtimestamp(ts, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+class ControlLoop:
+    """Senses conditions and remediates them through the governed spine, once per cooldown.
+
+    sense() -> list[condition]; each condition is a dict with a stable "key".
+    to_workload(condition) -> (workload, policy, cost) for the remediation.
+    spine_kwargs: the run_spine keyword args held constant across ticks — registry, binding,
+      capability, attestation, constraints, signer, verifier, and optionally admission / apply.
+    """
+
+    def __init__(self, *, sense, to_workload, spine_kwargs: dict, cooldown_s: float = 300.0,
+                 clock=time.time):
+        self._sense = sense
+        self._to_workload = to_workload
+        self._spine = spine_kwargs
+        self._cooldown = float(cooldown_s)
+        self._clock = clock
+        self._acted: dict[str, float] = {}
+        self._ledger: list[dict] = []
+
+    def tick(self) -> list[dict]:
+        """One sense→act pass. Returns the per-condition outcomes for this tick."""
+        now = self._clock()
+        out = []
+        for cond in self._sense():
+            key = cond.get("key")
+            last = self._acted.get(key)
+            if last is not None and now - last < self._cooldown:
+                out.append({"key": key, "outcome": "suppressed",
+                            "reason": "within cooldown", "at": _iso(now)})
+                continue
+            try:
+                workload, policy, cost = self._to_workload(cond)
+                result = ex.run_spine(workload, policy, cost=cost, **self._spine)
+                status = result.get("status", "unknown")
+            except Exception as exc:  # a bad remediation must not kill the loop
+                result, status = {}, "error"
+                err = str(exc)
+            else:
+                err = None
+            self._acted[key] = now
+            rec = {"key": key, "outcome": status, "backend": result.get("backend"),
+                   "grant_id": result.get("grant_id"), "condition": cond, "at": _iso(now)}
+            if err is not None:
+                rec["error"] = err
+            rec["receipt_digest"] = _seal({k: v for k, v in rec.items() if k != "receipt_digest"})
+            self._ledger.append(rec)
+            out.append(rec)
+        return out
+
+    def run(self, *, max_ticks: int, interval_s: float = 0.0) -> list[dict]:
+        for _ in range(max_ticks):
+            self.tick()
+            if interval_s:
+                time.sleep(interval_s)
+        return self._ledger
+
+    def ledger(self) -> list[dict]:
+        return list(self._ledger)
+
+
+if __name__ == "__main__":
+    # demo: a sensor that flags a backend as unhealthy; the loop remediates via a probe workload,
+    # then suppresses the same condition on the next tick (cooldown).
+    import mcp_a2a_grant as g
+    import mesh_telemetry as mt
+
+    key = b"loop-demo-key"
+    reg = mt.MeshRegistry()
+    reg.heartbeat("k8s-a", "k8s", 8)
+    aum = "sha256:" + "ab" * 32
+
+    loop = ControlLoop(
+        sense=lambda: [{"key": "unhealthy:paymentsvc", "target": "paymentsvc"}],
+        to_workload=lambda c: ({"name": "probe", "sensitivity": "normal", "scalable": False,
+                                "needs_gpu": False, "effect": "exec", "command": "echo probing"},
+                               {}, 1.0),
+        spine_kwargs=dict(
+            registry=reg,
+            binding={"spiffe_id": "spiffe://sourceos/agent/healer", "aum_digest": aum, "session_id": "sess_loop01"},
+            capability={"kind": "mcp_tool", "capability_ref": "capd://caps.compute.mesh-plane",
+                        "capability_digest": "sha256:" + "cd" * 32, "effect": "exec"},
+            attestation=g.attestation_bundle(spiffe_id="spiffe://sourceos/agent/healer", aum_digest=aum,
+                                             tpm_valid=True, cosign_valid=True),
+            constraints={"ops_allow": ["exec.run"]},
+            signer=g.hmac_signer(key), verifier=g.hmac_verifier(key)),
+        cooldown_s=300.0)
+    print("tick 1:", [(r["key"], r["outcome"]) for r in loop.tick()])
+    print("tick 2:", [(r["key"], r["outcome"]) for r in loop.tick()])

tools/test_control_loop.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Tests for the governed autonomous control loop. The safety properties are what matter: it acts
+through the governed spine, decides each condition once per cooldown, records fail-closed
+block/deny without acting ungoverned, and never lets one bad condition kill the loop."""
+import control_loop as cl
+import mcp_a2a_grant as g
+import mesh_telemetry as mt
+
+KEY = b"loop-test-key"
+AUM = "sha256:" + "ab" * 32
+
+
+class Clock:
+    def __init__(self, t=1000.0):
+        self.t = t
+
+    def __call__(self):
+        return self.t
+
+
+def _spine(reg, admission=None):
+    kw = dict(
+        registry=reg,
+        binding={"spiffe_id": "spiffe://sourceos/agent/healer", "aum_digest": AUM, "session_id": "sess_loop1"},
+        capability={"kind": "mcp_tool", "capability_ref": "capd://caps.x",
+                    "capability_digest": "sha256:" + "cd" * 32, "effect": "exec"},
+        attestation=g.attestation_bundle(spiffe_id="spiffe://sourceos/agent/healer", aum_digest=AUM,
+                                         tpm_valid=True, cosign_valid=True),
+        constraints={"ops_allow": ["exec.run"]},
+        signer=g.hmac_signer(KEY), verifier=g.hmac_verifier(KEY))
+    if admission is not None:
+        kw["admission"] = admission
+    return kw
+
+
+def _wl(effect="exec", gpu=False, sensitivity="normal"):
+    return {"name": "probe", "sensitivity": sensitivity, "scalable": gpu, "needs_gpu": gpu,
+            "effect": effect, "command": "echo hi"}
+
+
+def test_tick_remediates_a_firing_condition_through_the_governed_spine():
+    reg = mt.MeshRegistry()
+    reg.heartbeat("k", "k8s", 8)
+    loop = cl.ControlLoop(sense=lambda: [{"key": "unhealthy:svc"}],
+                          to_workload=lambda c: (_wl(), {}, 1.0), spine_kwargs=_spine(reg), clock=Clock())
+    out = loop.tick()
+    assert len(out) == 1 and out[0]["outcome"] == "ran" and out[0]["grant_id"]
+    assert out[0]["receipt_digest"].startswith("sha256:")
+
+
+def test_a_condition_is_decided_once_per_cooldown():
+    reg = mt.MeshRegistry()
+    reg.heartbeat("k", "k8s", 8)
+    clk = Clock(1000.0)
+    loop = cl.ControlLoop(sense=lambda: [{"key": "unhealthy:svc"}],
+                          to_workload=lambda c: (_wl(), {}, 1.0), spine_kwargs=_spine(reg),
+                          cooldown_s=300, clock=clk)
+    assert loop.tick()[0]["outcome"] == "ran"
+    clk.t = 1100.0                                  # within cooldown
+    assert loop.tick()[0]["outcome"] == "suppressed"
+    clk.t = 1400.0                                  # past the 300s cooldown
+    assert loop.tick()[0]["outcome"] == "ran"
+    assert len([r for r in loop.ledger() if r["outcome"] == "ran"]) == 2  # suppressed not ledgered
+
+
+def test_fail_closed_block_is_recorded_and_the_loop_continues():
+    reg = mt.MeshRegistry()
+    reg.heartbeat("v", "volunteer-boinc", 500)      # only an untrusted backend is up
+    loop = cl.ControlLoop(sense=lambda: [{"key": "c1"}, {"key": "c2"}],
+                          to_workload=lambda c: (_wl(sensitivity="sensitive"), {}, 1.0),
+                          spine_kwargs=_spine(reg), clock=Clock())
+    out = loop.tick()
+    assert [r["outcome"] for r in out] == ["blocked", "blocked"]  # both blocked, neither crashed
+    assert all(r["grant_id"] is None for r in out)                # nothing was granted
+
+
+def test_fail_closed_quota_denies_the_second_gpu_remediation():
+    import admission as adm
+    reg = mt.MeshRegistry()
+    reg.heartbeat("s", "hpc-slurm", 100)
+    ac = adm.AdmissionController({"spiffe://sourceos/agent/healer": {"gpu_max": 1}})
+    loop = cl.ControlLoop(sense=lambda: [{"key": "g1"}, {"key": "g2"}],
+                          to_workload=lambda c: (_wl(effect="exec", gpu=True), {}, 1.0),
+                          spine_kwargs=_spine(reg, admission=ac), clock=Clock())
+    assert [r["outcome"] for r in loop.tick()] == ["ran", "denied"]
+
+
+def test_a_raising_remediation_does_not_kill_the_loop():
+    reg = mt.MeshRegistry()
+    reg.heartbeat("k", "k8s", 8)
+
+    def to_wl(c):
+        if c["key"] == "bad":
+            raise ValueError("boom")
+        return (_wl(), {}, 1.0)
+
+    loop = cl.ControlLoop(sense=lambda: [{"key": "bad"}, {"key": "good"}],
+                          to_workload=to_wl, spine_kwargs=_spine(reg), clock=Clock())
+    out = loop.tick()
+    assert out[0]["outcome"] == "error" and "boom" in out[0]["error"]
+    assert out[1]["outcome"] == "ran"               # loop carried on to the next condition
+
+
+def test_run_executes_multiple_ticks():
+    reg = mt.MeshRegistry()
+    reg.heartbeat("k", "k8s", 8)
+    seen = [0]
+
+    def sense():
+        seen[0] += 1
+        return [{"key": f"c{seen[0]}"}]              # a fresh condition each tick
+
+    loop = cl.ControlLoop(sense=sense, to_workload=lambda c: (_wl(), {}, 1.0),
+                          spine_kwargs=_spine(reg), clock=Clock())
+    ledger = loop.run(max_ticks=3)
+    assert len([r for r in ledger if r["outcome"] == "ran"]) == 3
+
+
+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)} control-loop tests passed")
+    sys.exit(0)

tools/validate.py

@@ -22,6 +22,7 @@
     "capd/devspace.local-dev.capd.json",
     "capd/cloudshell-fog.capd.json",
     "capd/knowledge-commons.mesh.capd.json",
+    "capd/self-healing-loop.mesh.capd.json",
     "tools/promotion_gate.py",
     "tools/portal_server.py",
     "tools/compute_plane.py",
@@ -32,6 +33,7 @@
     "tools/executor.py",
     "tools/sourceosctl.py",
     "tools/admission.py",
+    "tools/control_loop.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.
@@ -40,6 +42,7 @@
     "capd/devspace.local-dev.capd.json": "caps.dev.devspace-inner-loop",
     "capd/cloudshell-fog.capd.json": "caps.compute.cloudshell-fog",
     "capd/knowledge-commons.mesh.capd.json": "caps.knowledge.commons",
+    "capd/self-healing-loop.mesh.capd.json": "caps.compute.self-healing-loop",
 }
 
 errors: list[str] = []