feat(mesh): pull/lease streaming scheduler — Pando StreamLender/Limiter (move 2)

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 verify
+.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 lease
 
 validate: ## repo hygiene + CapD validity
 	python3 tools/validate.py
@@ -39,6 +39,9 @@ loop: ## autonomous control loop demo: sense -> governed spine -> act, once per
 verify: ## volunteer-mesh verification demo: redundant quorum over untrusted worker results
 	cd tools && python3 work_unit.py
 
+lease: ## pull/lease scheduler demo: workers pull WUs, crash-stop re-lending, ordered re-merge
+	cd tools && python3 lease_scheduler.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

@@ -6,6 +6,7 @@
   "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",
+    "scheduler": "tools/lease_scheduler.py",
     "executor": "tools/executor.py",
     "placement": "tools/compute_plane.py",
     "grant_authority": "tools/mcp_a2a_grant.py",

docs/VOLUNTEER_MESH.md

@@ -90,7 +90,7 @@ results back.
 | Governed dispatch of a WU | `executor` adapters (local/k8s/slurm/wasm/descriptor/connector) | ✅ |
 | Identity/attestation (move 4) | `mcp_a2a_grant` (SPIFFE binding + TPM/cosign attestation) | ✅ |
 | Quotas/admission | `admission.py` (additive; make credits additive-only per move 4) | ✅ |
-| Pull/lease streaming dispatch (move 2) | StreamLender + Limiter; heartbeat liveness (`mesh_telemetry` is the base) | ○ next |
+| **Pull/lease streaming dispatch (move 2)** | `lease_scheduler.py` — StreamLender + Limiter: workers pull/lease when idle (adaptive, no speed estimation), conservative single-copy, crash-stop re-lending by index, ordered re-merge, heartbeat liveness | ✅ (this change) |
 | Split data plane + DataBridge (move 3) | signed-URL nearest-endpoint upload + async replication + verify-download barrier | ○ next |
 | Instrumented signals + refusal (move 5) | wrap reputation/confidence as `{value,uncertainty,provenance,validity,refusal}` | ○ next |
 | Domain-scoped reputation (move 7) | per-`(worker,domain)` card on `work_unit.Reputation` | ○ next |

tools/lease_scheduler.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""Pull/lease streaming dispatch — Pando's StreamLender + Limiter, governed.
+
+Push WU-assignment is brittle on a churny volunteer mesh: you'd estimate each worker's speed and
+re-push on every crash. The elegant alternative is PULL: a worker LEASES work when it is idle, so
+faster/idler devices simply pull more (adaptive, no speed estimation) and slow ones pull less
+(automatic backpressure). Properties:
+
+  * conservative — one copy of a Work Unit to at most one worker at a time, maximizing DISTINCT
+    units in flight (redundant-quorum is an optional overlay layered on top, not the default).
+  * crash-stop work-stealing — a worker that misses its lease deadline (crashed/left) has its
+    borrowed unit RE-LENT by index to someone else; nothing is lost.
+  * ordered — results re-merge by index for determinism.
+  * bounded — a Limiter caps in-flight leases per worker.
+
+Heartbeat extends a lease's deadline (progress signal); silence past the deadline reclaims it.
+"""
+from __future__ import annotations
+
+import time
+
+
+class LeaseScheduler:
+    def __init__(self, work_units, *, max_in_flight: int = 4, lease_ttl_s: float = 30.0,
+                 clock=time.time):
+        self._pending: list[tuple[int, object]] = list(enumerate(work_units))  # ordered by index
+        self._leased: dict[str, dict] = {}
+        self._done: dict[int, object] = {}
+        self._inflight: dict[str, set] = {}
+        self._max = int(max_in_flight)
+        self._ttl = float(lease_ttl_s)
+        self._clock = clock
+        self._seq = 0
+
+    def _reclaim(self, now: float) -> list[str]:
+        """Return WUs whose lease deadline passed (worker crashed/left) to the pool — re-lent by
+        index. A unit already completed is not re-lent."""
+        expired = [lid for lid, l in self._leased.items() if now > l["deadline"]]
+        for lid in expired:
+            l = self._leased.pop(lid)
+            self._inflight.get(l["worker"], set()).discard(lid)
+            if l["index"] not in self._done:
+                self._pending.append((l["index"], l["wu"]))
+        if expired:
+            self._pending.sort(key=lambda t: t[0])  # keep ordered for deterministic re-merge
+        return expired
+
+    def lease(self, worker: str, n: int = 1) -> list[dict]:
+        """A worker pulls up to n Work Units (idle workers call this more → adaptive). Respects the
+        per-worker Limiter."""
+        now = self._clock()
+        self._reclaim(now)
+        held = self._inflight.setdefault(worker, set())
+        out = []
+        while self._pending and len(held) < self._max and len(out) < n:
+            index, wu = self._pending.pop(0)
+            self._seq += 1
+            lid = f"lease-{self._seq}"
+            self._leased[lid] = {"index": index, "wu": wu, "worker": worker, "deadline": now + self._ttl}
+            held.add(lid)
+            out.append({"lease_id": lid, "index": index, "wu": wu})
+        return out
+
+    def complete(self, lease_id: str, output) -> bool:
+        """Return a result. False if the lease was already reclaimed (a slow/crashed worker whose
+        unit was re-lent) — its late result is dropped, the re-lent copy is authoritative."""
+        l = self._leased.pop(lease_id, None)
+        if l is None:
+            return False
+        self._inflight.get(l["worker"], set()).discard(lease_id)
+        self._done.setdefault(l["index"], output)
+        return True
+
+    def heartbeat(self, worker: str, lease_id: str | None = None) -> None:
+        """Extend the deadline of a worker's lease(s) — a progress signal that keeps it from being
+        reclaimed."""
+        now = self._clock()
+        for lid, l in self._leased.items():
+            if l["worker"] == worker and (lease_id is None or lid == lease_id):
+                l["deadline"] = now + self._ttl
+
+    def progress(self) -> dict:
+        return {"pending": len(self._pending), "leased": len(self._leased), "done": len(self._done)}
+
+    def drained(self) -> bool:
+        self._reclaim(self._clock())
+        return not self._pending and not self._leased
+
+    def results_in_order(self) -> list:
+        return [self._done[i] for i in sorted(self._done)]
+
+
+if __name__ == "__main__":
+    import json
+
+    class Clock:
+        def __init__(self, t=0.0):
+            self.t = t
+
+        def __call__(self):
+            return self.t
+
+    clk = Clock()
+    sched = LeaseScheduler([f"wu{i}" for i in range(5)], max_in_flight=2, lease_ttl_s=10, clock=clk)
+    # fast worker A pulls + completes; worker B leases one then "crashes" (never completes).
+    a1 = sched.lease("A", 1)[0]
+    b1 = sched.lease("B", 1)[0]
+    sched.complete(a1["lease_id"], a1["wu"] + "-done-by-A")
+    clk.t = 20  # B missed its deadline -> its unit is re-lent
+    a2 = sched.lease("A", 1)  # picks up B's abandoned unit + more
+    for lz in a2:
+        sched.complete(lz["lease_id"], lz["wu"] + "-done-by-A")
+    while not sched.drained():
+        for lz in sched.lease("A", 4):
+            sched.complete(lz["lease_id"], lz["wu"] + "-done-by-A")
+    print(json.dumps({"drained": sched.drained(), "results": sched.results_in_order()}, indent=2))

tools/test_lease_scheduler.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Tests for the pull/lease scheduler. Load-bearing: conservative single-copy dispatch (one WU to
+one worker), crash-stop re-lending (a dead worker's unit is not lost), ordered re-merge, and the
+per-worker Limiter. Time is injected so crash detection is deterministic."""
+import lease_scheduler as ls
+
+
+class Clock:
+    def __init__(self, t=0.0):
+        self.t = t
+
+    def __call__(self):
+        return self.t
+
+
+def test_limiter_caps_in_flight_per_worker():
+    s = ls.LeaseScheduler(range(10), max_in_flight=3, clock=Clock())
+    assert len(s.lease("w", n=10)) == 3
+
+
+def test_conservative_single_copy_no_double_lease():
+    s = ls.LeaseScheduler(["a", "b"], max_in_flight=4, clock=Clock())
+    w1 = s.lease("w1", 2)
+    assert len(w1) == 2 and s.lease("w2", 2) == []  # both held by w1; nothing for w2
+
+
+def test_results_re_merge_in_index_order():
+    s = ls.LeaseScheduler(["a", "b", "c"], max_in_flight=4, clock=Clock())
+    got = s.lease("w", 3)
+    s.complete(got[2]["lease_id"], "C")  # complete out of order
+    s.complete(got[0]["lease_id"], "A")
+    s.complete(got[1]["lease_id"], "B")
+    assert s.results_in_order() == ["A", "B", "C"]
+
+
+def test_crashed_worker_unit_is_relent_and_late_result_dropped():
+    clk = Clock()
+    s = ls.LeaseScheduler(["only"], max_in_flight=4, lease_ttl_s=10, clock=clk)
+    a = s.lease("A", 1)[0]
+    assert s.lease("B", 1) == []          # conservative: B gets nothing while A holds it
+    clk.t = 20                            # A missed its deadline -> crashed
+    b = s.lease("B", 1)
+    assert len(b) == 1 and b[0]["wu"] == "only"   # re-lent to B
+    assert s.complete(a["lease_id"], "stale") is False   # A's late result is dropped
+    assert s.complete(b[0]["lease_id"], "fresh") is True
+    assert s.results_in_order() == ["fresh"]      # the unit ran exactly once, authoritative copy
+
+
+def test_heartbeat_prevents_reclaim():
+    clk = Clock()
+    s = ls.LeaseScheduler(["x"], lease_ttl_s=10, clock=clk)
+    s.lease("A", 1)
+    clk.t = 8
+    s.heartbeat("A")          # extends deadline to 18
+    clk.t = 15
+    assert s.lease("B", 1) == []   # still alive; not reclaimed
+
+
+def test_adaptive_a_fast_worker_drains_more():
+    s = ls.LeaseScheduler(range(6), max_in_flight=2, clock=Clock())
+    a_total = 0
+    while not s.drained():
+        got = s.lease("A", 2)   # A keeps pulling; B never does
+        for lz in got:
+            s.complete(lz["lease_id"], lz["index"])
+            a_total += 1
+    assert a_total == 6
+
+
+def test_end_to_end_two_workers_one_crash_each_wu_once():
+    clk = Clock()
+    s = ls.LeaseScheduler([f"wu{i}" for i in range(4)], max_in_flight=1, lease_ttl_s=10, clock=clk)
+    a = s.lease("A", 1)[0]
+    s.lease("B", 1)                    # B takes one then crashes (never completes)
+    s.complete(a["lease_id"], a["wu"])
+    clk.t = 20                         # B reclaimed
+    while not s.drained():
+        for lz in s.lease("A", 4):
+            s.complete(lz["lease_id"], lz["wu"])
+    assert sorted(s.results_in_order()) == ["wu0", "wu1", "wu2", "wu3"]  # every WU exactly once
+
+
+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)} lease-scheduler tests passed")
+    sys.exit(0)

tools/validate.py

@@ -37,6 +37,7 @@
     "tools/control_loop.py",
     "tools/devspace.py",
     "tools/work_unit.py",
+    "tools/lease_scheduler.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.

tools/work_unit.py

@@ -16,6 +16,9 @@
 
 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.
+
+Dispatch of Work Units to workers is pull/lease (see `lease_scheduler.py`): verification here is the
+optional overlay layered on top, keyed on stakes x reputation.
 """
 from __future__ import annotations