diff --git a/Makefile b/Makefile index 4d10439..654c17e 100644 --- a/Makefile +++ b/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" diff --git a/capd/self-healing-loop.mesh.capd.json b/capd/self-healing-loop.mesh.capd.json new file mode 100644 index 0000000..7dc599e --- /dev/null +++ b/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 + } +} diff --git a/tools/control_loop.py b/tools/control_loop.py new file mode 100644 index 0000000..aff0ea3 --- /dev/null +++ b/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()]) diff --git a/tools/test_control_loop.py b/tools/test_control_loop.py new file mode 100644 index 0000000..3f0bfbe --- /dev/null +++ b/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) diff --git a/tools/validate.py b/tools/validate.py index d40ef18..66710e7 100644 --- a/tools/validate.py +++ b/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] = []