feat(work-unit): volunteer-mesh verification — trust untrusted results by quorum

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 loop
+.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 verify
 
 validate: ## repo hygiene + CapD validity
 	python3 tools/validate.py
@@ -36,6 +36,9 @@ run: ## sourceosctl: run a workload governed across the mesh (e.g. make run ARGS
 loop: ## autonomous control loop demo: sense -> governed spine -> act, once per cooldown
 	cd tools && python3 control_loop.py
 
+verify: ## volunteer-mesh verification demo: redundant quorum over untrusted worker results
+	cd tools && python3 work_unit.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/volunteer-mesh-verification.mesh.capd.json

@@ -0,0 +1,30 @@
+{
+  "capability_id": "caps.compute.volunteer-mesh-verification@0.1.0",
+  "kind": "compute.volunteer-verification",
+  "status": "experimental",
+  "name": "Volunteer-mesh verification — trust untrusted results by quorum",
+  "description": "How a Folding@home-scale volunteer grid (hundreds of thousands of anonymous, churny, possibly-malicious workers) becomes trustworthy. A Grant proves WHO ran a Work Unit; it cannot prove the RESULT is correct. This plane verifies results fail-closed: a Work Unit is run redundantly on N independent workers and only a result a quorum agrees on (identical output digest) is accepted; stochastic tasks use spot-check canaries with known answers; reliable backends (a cluster, per CluBORun) can stand in as reference verifiers. Reputation is a per-worker verified-success record that weights allocation and routes around bad actors. Implements the Dual-Orchestration PROOF_MODE (redundant | spot_check | tee | zk | optimistic).",
+  "links": {
+    "engine": "tools/work_unit.py",
+    "executor": "tools/executor.py",
+    "placement": "tools/compute_plane.py",
+    "grant_authority": "tools/mcp_a2a_grant.py",
+    "ledger": "artifacts/gate-decisions + mcp-receipts (sealed)",
+    "reference_pattern": "BOINC / Folding@home redundant verification + Science United matchmaking + CluBORun reference-cluster; Dual-Orchestration Model PROOF_MODE"
+  },
+  "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": {
+    "result_verification": true,
+    "fail_closed": true,
+    "redundant_quorum": true,
+    "spot_check": true,
+    "reputation_weighted": true,
+    "sensitive_never_untrusted": true,
+    "evidence_emitting": true
+  }
+}

tools/test_work_unit.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""Tests for the Work Unit + verification plane. The load-bearing property: an untrusted worker's
+result is NEVER accepted unless a quorum of independent workers agrees on it — and liars lose
+reputation. This is what makes a 400k-node volunteer mesh trustworthy."""
+import work_unit as wu
+
+
+def test_work_unit_is_content_addressed():
+    a = wu.mint_work_unit(task="t", params={"k": 1})
+    b = wu.mint_work_unit(task="t", params={"k": 1})
+    c = wu.mint_work_unit(task="t", params={"k": 2})
+    assert a["wu_id"] == b["wu_id"] and a["wu_id"] != c["wu_id"]
+    assert a["wu_id"].startswith("wu:")
+
+
+def test_replication_defaults_by_proof_mode():
+    assert wu.mint_work_unit(task="t", proof_mode="redundant")["replication"] == 3
+    assert wu.mint_work_unit(task="t", proof_mode="spot_check")["replication"] == 1
+
+
+def test_quorum_accepts_the_majority_and_flags_the_liar():
+    unit = wu.mint_work_unit(task="image.infer", proof_mode="redundant")
+    results = [{"worker": "a", "output": {"label": "cat"}},
+               {"worker": "b", "output": {"label": "cat"}},
+               {"worker": "c", "output": {"label": "GARBAGE"}}]  # the liar
+    v = wu.Verifier().redundant_quorum(unit, results)
+    assert v["verified"] is True
+    assert v["workers_agree"] == ["a", "b"] and v["workers_disagree"] == ["c"]
+    assert v["accepted_digest"] == wu.digest({"label": "cat"})
+
+
+def test_no_quorum_is_fail_closed():
+    unit = wu.mint_work_unit(task="t", proof_mode="redundant")  # replication 3, threshold 2
+    results = [{"worker": "a", "output": 1}, {"worker": "b", "output": 2}, {"worker": "c", "output": 3}]
+    v = wu.Verifier().redundant_quorum(unit, results)
+    assert v["verified"] is False and v["accepted_digest"] is None
+    assert "no quorum" in v["reason"]
+
+
+def test_a_tie_below_threshold_does_not_verify():
+    unit = wu.mint_work_unit(task="t", proof_mode="redundant", replication=4)  # threshold 3
+    results = [{"worker": "a", "output": 1}, {"worker": "b", "output": 1},
+               {"worker": "c", "output": 2}, {"worker": "d", "output": 2}]  # 2-2, neither reaches 3
+    assert wu.Verifier().redundant_quorum(unit, results)["verified"] is False
+
+
+def test_spot_check_rejects_a_wrong_canary():
+    good = wu.Verifier().spot_check({"worker": "a", "output": {"ans": 42}}, {"ans": 42})
+    bad = wu.Verifier().spot_check({"worker": "b", "output": {"ans": 7}}, {"ans": 42})
+    assert good["verified"] is True and bad["verified"] is False
+
+
+def test_reputation_rewards_agreement_and_penalizes_liars():
+    rep = wu.Reputation()
+    rep.record(agreed=["a", "b"], disagreed=["c"])
+    rep.record(agreed=["a", "b"], disagreed=["c"])
+    assert rep.score("a") == 1.0 and rep.score("c") == 0.0
+    assert rep.trusted("a") is False  # only 2 samples < min_samples 3
+    rep.record(agreed=["a"], disagreed=[])
+    assert rep.trusted("a") is True   # 3 verified/3 total >= 0.8
+    assert rep.trusted("c") is False
+
+
+def test_verify_and_score_ties_verification_to_reputation():
+    unit = wu.mint_work_unit(task="t", proof_mode="redundant")
+    rep = wu.Reputation()
+    results = [{"worker": "a", "output": "x"}, {"worker": "b", "output": "x"},
+               {"worker": "evil", "output": "lie"}]
+    verdict = wu.verify_and_score(unit, results, rep)
+    assert verdict["verified"] is True
+    assert rep.score("a") == 1.0 and rep.score("evil") == 0.0  # the liar earned nothing
+
+
+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)} work-unit tests passed")
+    sys.exit(0)

tools/validate.py

@@ -23,6 +23,7 @@
     "capd/cloudshell-fog.capd.json",
     "capd/knowledge-commons.mesh.capd.json",
     "capd/self-healing-loop.mesh.capd.json",
+    "capd/volunteer-mesh-verification.mesh.capd.json",
     "tools/promotion_gate.py",
     "tools/portal_server.py",
     "tools/compute_plane.py",
@@ -35,6 +36,7 @@
     "tools/admission.py",
     "tools/control_loop.py",
     "tools/devspace.py",
+    "tools/work_unit.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.
@@ -44,6 +46,7 @@
     "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",
+    "capd/volunteer-mesh-verification.mesh.capd.json": "caps.compute.volunteer-mesh-verification",
 }
 
 errors: list[str] = []

tools/work_unit.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""Work Unit + verification plane — how an UNTRUSTED volunteer mesh becomes trustworthy.
+
+A Grant proves WHO ran a Work Unit; it cannot prove the RESULT is correct. On a Folding@home-scale
+grid (hundreds of thousands of anonymous volunteers) some workers are broken, slow, or malicious —
+they return garbage or lie. Volunteer computing solves this by VERIFYING results, and that is the
+piece our compute plane was missing.
+
+This implements the Dual-Orchestration `PROOF_MODE` verification, fail-closed:
+
+  * redundant quorum — run the same Work Unit on N independent workers; accept only a result a
+    quorum agrees on (identical output digest). No quorum -> rejected, never accepted unverified.
+  * spot-check — embed a canary sub-task with a known answer; a worker that gets the canary wrong is
+    rejected and loses reputation (for stochastic tasks where bit-exact quorum doesn't apply).
+  * a reliable backend (a cluster, per CluBORun) can stand in as a reference verifier.
+
+Reputation is a per-worker moving record of verified successes; it weights future allocation and is
+how the mesh routes around bad actors without trusting any single node.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+
+# Default replication (independent runs) required to VERIFY a result, per proof mode.
+REPLICATION = {"redundant": 3, "spot_check": 1, "tee": 1, "zk": 1, "optimistic": 1}
+
+
+def _canon(obj) -> str:
+    return "sha256:" + hashlib.sha256(
+        json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
+
+
+def digest(output) -> str:
+    """Content digest of a worker's output — what quorum agreement is computed over."""
+    return _canon(output)
+
+
+def mint_work_unit(*, task: str, inputs: list | None = None, params: dict | None = None,
+                   proof_mode: str = "redundant", sandbox: str = "wasm", ttl_s: int = 60,
+                   replication: int | None = None) -> dict:
+    """A content-addressed Work Unit. `wu_id` is derived from the task definition, so the same work
+    has the same id (dedup + reproducibility)."""
+    body = {"task": task, "inputs": inputs or [], "params": params or {},
+            "proof_mode": proof_mode, "sandbox": sandbox}
+    return {"wu_id": "wu:" + digest(body).split(":", 1)[1][:16],
+            **body, "ttl_s": ttl_s,
+            "replication": replication or REPLICATION.get(proof_mode, 1)}
+
+
+class Verifier:
+    """Fail-closed result verification for untrusted workers."""
+
+    def redundant_quorum(self, wu: dict, results: list, *, threshold: int | None = None) -> dict:
+        """results: [{worker, output}] from independent runs. Accept the output a majority agrees on
+        (by digest). threshold defaults to a strict majority of the WU's replication factor."""
+        n = wu.get("replication", len(results))
+        threshold = threshold or (n // 2 + 1)
+        groups: dict[str, list] = {}
+        for r in results:
+            groups.setdefault(digest(r["output"]), []).append(r["worker"])
+        if not groups:
+            return {"verified": False, "accepted_digest": None, "reason": "no results", "threshold": threshold}
+        winner, agree = max(groups.items(), key=lambda kv: len(kv[1]))
+        verified = len(agree) >= threshold
+        disagree = [w for d, ws in groups.items() if d != winner for w in ws]
+        return {"verified": verified,
+                "accepted_digest": winner if verified else None,
+                "workers_agree": sorted(agree),
+                "workers_disagree": sorted(disagree),
+                "threshold": threshold, "received": len(results),
+                "reason": ("quorum reached" if verified
+                           else f"no quorum: best {len(agree)}/{threshold} agreed")}
+
+    def spot_check(self, result: dict, expected_output) -> dict:
+        """A canary with a known answer — for stochastic tasks. Wrong canary -> rejected."""
+        ok = digest(result.get("output")) == digest(expected_output)
+        return {"verified": ok, "worker": result.get("worker"),
+                "reason": "canary matched" if ok else "canary FAILED — worker output rejected"}
+
+
+class Reputation:
+    """Per-worker verified-success record; weights allocation and routes around bad actors."""
+
+    def __init__(self):
+        self._rep: dict[str, dict] = {}
+
+    def _r(self, worker: str) -> dict:
+        return self._rep.setdefault(worker, {"verified": 0, "total": 0})
+
+    def record(self, *, agreed: list, disagreed: list) -> None:
+        for w in agreed:
+            r = self._r(w)
+            r["verified"] += 1
+            r["total"] += 1
+        for w in disagreed:
+            self._r(w)["total"] += 1
+
+    def score(self, worker: str) -> float:
+        r = self._r(worker)
+        return round(r["verified"] / r["total"], 3) if r["total"] else 0.0
+
+    def trusted(self, worker: str, *, minimum: float = 0.8, min_samples: int = 3) -> bool:
+        r = self._r(worker)
+        return r["total"] >= min_samples and self.score(worker) >= minimum
+
+
+def verify_and_score(wu: dict, results: list, reputation: Reputation | None = None,
+                     verifier: Verifier | None = None) -> dict:
+    """Convenience: quorum-verify a WU's results and update reputation. Returns the verdict."""
+    verifier = verifier or Verifier()
+    verdict = verifier.redundant_quorum(wu, results)
+    if reputation is not None and verdict.get("workers_agree") is not None:
+        reputation.record(agreed=verdict.get("workers_agree", []),
+                          disagreed=verdict.get("workers_disagree", []))
+    return verdict
+
+
+if __name__ == "__main__":
+    wu = mint_work_unit(task="image.infer@v3", params={"top_k": 3}, proof_mode="redundant")
+    # 3 workers: two agree, one lies.
+    results = [{"worker": "vol-a", "output": {"label": "cat", "score": 0.9}},
+               {"worker": "vol-b", "output": {"label": "cat", "score": 0.9}},
+               {"worker": "vol-c", "output": {"label": "GARBAGE"}}]
+    rep = Reputation()
+    verdict = verify_and_score(wu, results, rep)
+    print(json.dumps({"wu": wu["wu_id"], "verified": verdict["verified"],
+                      "agree": verdict["workers_agree"], "disagree": verdict["workers_disagree"],
+                      "rep_vol_a": rep.score("vol-a"), "rep_vol_c": rep.score("vol-c")}, indent=2))