From 8ea14e24b5b091857ea8010999c5528257730882 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Tue, 4 Aug 2026 00:51:54 -0400 Subject: [PATCH] feat(executor): parallel/MPI jobs + real SLURM sbatch + connector-as-dispatch (IBM PE codified) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers "what job types", "how do we leverage SLURM", and "the MCP-connector pattern" — one Grant-gated executor now dispatches batch, parallel, stateful, and connector work alike. - Parallel / MPI (the IBM Parallel Environment POE pattern): K8sAdapter emits an Indexed Job (completions/parallelism=N, completionMode=Indexed) — each task gets JOB_COMPLETION_INDEX as its rank. `sourceosctl run --parallelism N --nodes M`. - Real SLURM: SlurmAdapter emits a real sbatch script (#SBATCH --ntasks/--nodes/--gres, `srun` to launch MPI ranks, grant-bound via --comment), and submits it via `ssh $SOURCEOS_SLURM_LOGIN sbatch` when configured (else emits the script). Replaces the hpc-slurm descriptor stub. - Connector-as-dispatch: ConnectorAdapter + a `connector` backend (external/untrusted, so sensitive data won't auto-route to a vendor). A Gemini/OpenAI/Claude Files-API / MCP call is the SAME materialize->handle->dispatch->result shape as a compute job, gated by the same Grant. Also fixes a latent bug: `place` reused cmd_run but its parser omits the run-only flags (--cost/--inception/--parallelism) -> AttributeError; cmd_run now reads them defensively. Tests: +1 indexed-job +1 sbatch +1 connector +1 connector-plane +1 place-regression = 110 green. docs/PATTERN_INTEGRATION.md updated (these three frontiers now shipped; volunteer-mesh is next). --- docs/PATTERN_INTEGRATION.md | 19 ++++++---- tools/compute_plane.py | 3 ++ tools/executor.py | 70 ++++++++++++++++++++++++++++++++++--- tools/sourceosctl.py | 16 ++++++--- tools/test_compute_plane.py | 10 ++++++ tools/test_executor.py | 43 ++++++++++++++++++++--- tools/test_sourceosctl.py | 6 ++++ 7 files changed, 145 insertions(+), 22 deletions(-) diff --git a/docs/PATTERN_INTEGRATION.md b/docs/PATTERN_INTEGRATION.md index 57d37a6..5d9c2f6 100644 --- a/docs/PATTERN_INTEGRATION.md +++ b/docs/PATTERN_INTEGRATION.md @@ -30,8 +30,8 @@ graded. `✅` shipped · `◑` partial · `○` gap (named next step). |---|---|---| | **Placement across substrates** (LoadLeveler/SLURM-style) | `compute_plane.place` over local/k8s/hpc-slurm/wasm/p2p/volunteer/blockchain | ✅ | | **Batch job** | `Job` (k8s, real) / `sbatch` (slurm) | ✅ k8s · ○ slurm submit | -| **Parallel / MPI job (N ranks)** — the POE pattern | Indexed `Job` (k8s) / `--ntasks` MPI + `--array` (slurm) | ○ **next unit** | -| **Real SLURM submission** | today: placement + descriptor only (`DescriptorAdapter`); real `sbatch`/`srun`/MPI adapter | ○ **next unit** | +| **Parallel / MPI job (N ranks)** — the POE pattern | `executor.K8sAdapter` emits an **Indexed Job** (completions/parallelism=N, `JOB_COMPLETION_INDEX` = rank); `sourceosctl run --parallelism N` | ✅ | +| **Real SLURM submission** | `executor.SlurmAdapter` emits a real `sbatch` script (`--ntasks` + `srun` MPI ranks), submits via `ssh sbatch` | ✅ | | **Edge/Fog K3s ↔ Cloud Twin sync** over intermittent links (LAN/WAN/sneakernet) | mesh telemetry + hyperswarm scale-up capability; real twin-sync + S3 export | ◑ | ## Reach, governance, evidence @@ -40,13 +40,18 @@ graded. `✅` shipped · `◑` partial · `○` gap (named next step). |---|---|---| | **SSH gateway to a device fleet** (ShellHub: Server + Agents on computer/device/container/server) | cloud-shell fog: Edge Gateway + HyperSwarm discovery + Grant-bound attach; a ShellHub-style agent per node | ◑ | | **7-layer PaaS** (UX→Object Store→Derived→Vendor→Retrieval→Policy→Tool-runtime) | knowledge commons (canonical + derived + provenance), MCP surface (tool runtime), promotion gate (policy) | ◑ | -| **Governed connector calls** (Gemini/OpenAI/Claude Files APIs — materialize→handle→dispatch→result) | the SAME governed dispatch as a compute job: a `connector` effect through grant + executor | ○ **next unit** | +| **Governed connector calls** (Gemini/OpenAI/Claude Files APIs — materialize→handle→dispatch→result) | `executor.ConnectorAdapter` + a `connector` backend (external/untrusted): a connector call is the SAME Grant-gated dispatch as a compute job | ✅ | | **Append-only audit ledger** (every diagram) | sealed receipts (`artifacts/`), MCP-A2A ledger conformance | ✅ | ## What this establishes The compute mesh + grants + admission is the **governance & scale-out substrate**; the DevSpace/ -Sandbox/StatefulSet plane is the **environment & stateful substrate**. The two open frontiers that -would make the HPC/connector story first-class are one unit: **parallel/MPI jobs (Indexed Job + -real SLURM) and connector-call-as-dispatch** — unifying "run a job" and "call a connector" under one -`Grant`-gated executor. That is the codification the IBM Parallel Environment pattern is asking for. +Sandbox/StatefulSet plane is the **environment & stateful substrate**; and one `Grant`-gated executor +now dispatches **batch, parallel/MPI (Indexed Job / SLURM `sbatch`+`srun`), stateful, and connector** +work alike — "run a job" and "call a connector" are the same governed dispatch. That codifies the +IBM Parallel Environment / HPC Toolkit pattern on the mesh. + +**Remaining frontier** (the next research-driven unit): the **volunteer-compute / global-mesh +substrate** — how a Folding@home-scale volunteer grid (untrusted, churny, 400k-node) plugs in +governed, per the dual-orchestration design and the volunteer-computing corpus. See the incoming +synthesis. diff --git a/tools/compute_plane.py b/tools/compute_plane.py index e7a0e1a..d568036 100644 --- a/tools/compute_plane.py +++ b/tools/compute_plane.py @@ -31,6 +31,9 @@ "p2p-mesh": {"kind": "hyperswarm", "trust": "untrusted", "elasticity": 8, "gpu": False}, "volunteer-boinc": {"kind": "volunteer", "trust": "untrusted", "elasticity": 10, "gpu": False}, "blockchain-rlc": {"kind": "compute-market", "trust": "untrusted", "elasticity": 8, "gpu": True}, + # A vendor connector (Gemini/OpenAI/Claude Files API, or an MCP tool) is external/untrusted — + # sensitive data does not auto-route here without explicit policy. + "connector": {"kind": "connector", "trust": "untrusted", "elasticity": 9, "gpu": True}, } diff --git a/tools/executor.py b/tools/executor.py index 6ed68cd..1a4f9ff 100644 --- a/tools/executor.py +++ b/tools/executor.py @@ -77,15 +77,23 @@ def manifest(self, workload, decision, grant): if pvc: container["volumeMounts"] = [{"name": "inception", "mountPath": "/var/lib/sourceos/inception"}] pod_spec["volumes"] = [{"name": "inception", "persistentVolumeClaim": {"claimName": pvc}}] + job_spec = {"backoffLimit": 0, "ttlSecondsAfterFinished": 3600, + "template": {"metadata": {"labels": {"sourceos.io/grant-id": grant["grant_id"]}}, + "spec": pod_spec}} + # Parallel / MPI job (the IBM Parallel Environment POE pattern): an Indexed Job runs N tasks, + # each getting JOB_COMPLETION_INDEX as its rank. + parallelism = int(workload.get("parallelism", 1)) + if parallelism > 1: + job_spec["parallelism"] = parallelism + job_spec["completions"] = int(workload.get("completions", parallelism)) + job_spec["completionMode"] = "Indexed" 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": pod_spec}}, + "spec": job_spec, } def dispatch(self, workload, decision, grant, *, apply): @@ -127,9 +135,61 @@ def dispatch(self, workload, decision, grant, *, apply): "note": f"hand this descriptor to the {self.backend} scheduler over a Grant-bound channel"} +class SlurmAdapter: + """HPC/SLURM: emits a real sbatch script (the IBM Parallel Environment POE pattern — N tasks via + --ntasks, launched with srun for MPI ranks) and submits it via `ssh sbatch` when a login + node is configured (SOURCEOS_SLURM_LOGIN), else emits the script.""" + backend = "hpc-slurm" + + def sbatch_script(self, workload, grant): + ntasks = int(workload.get("parallelism", 1)) + nodes = int(workload.get("nodes", 1)) + lines = ["#!/bin/bash", + f"#SBATCH --job-name={workload.get('name', 'sourceos')}", + f"#SBATCH --ntasks={ntasks}", + f"#SBATCH --nodes={nodes}", + f"#SBATCH --comment=grant:{grant['grant_id']}"] + if workload.get("needs_gpu"): + lines.append("#SBATCH --gres=gpu:1") + cmd = workload.get("command") or "true" + lines.append(f"srun {cmd}" if ntasks > 1 else cmd) # srun launches the N MPI ranks + return "\n".join(lines) + "\n" + + def dispatch(self, workload, decision, grant, *, apply): + import os + script = self.sbatch_script(workload, grant) + login = os.environ.get("SOURCEOS_SLURM_LOGIN") + if apply and login and shutil.which("ssh"): + proc = subprocess.run(["ssh", login, "sbatch"], input=script, + capture_output=True, text=True, timeout=60) + return {"kind": "hpc-slurm", "applied": proc.returncode == 0, "script": script, + "login": login, "sbatch": proc.stdout.strip() or proc.stderr.strip()} + return {"kind": "hpc-slurm", "applied": False, "script": script, + "note": "set SOURCEOS_SLURM_LOGIN to submit via `ssh sbatch`"} + + +class ConnectorAdapter: + """A remote connector call (Gemini/OpenAI/Claude Files API, or an MCP tool) as a GOVERNED + dispatch — the same materialize->handle->dispatch->result shape as a compute job, gated by the + same Grant. Emits the connector-call descriptor; the grant-bound connector runtime makes the + actual call.""" + backend = "connector" + + def dispatch(self, workload, decision, grant, *, apply): + return {"kind": "connector", "applied": False, + "call": {"connector": workload.get("connector", "mcp"), + "operation": workload.get("operation", "tools/call"), + "artifact_ref": workload.get("artifact_ref"), + "effect": grant["capability"].get("effect"), + "grant_id": grant["grant_id"], + "session": grant["binding"].get("session_id")}, + "note": "grant-bound connector call (materialize -> handle -> dispatch -> result)"} + + 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 = {LocalAdapter().backend: LocalAdapter(), K8sAdapter().backend: K8sAdapter(), + SlurmAdapter().backend: SlurmAdapter(), ConnectorAdapter().backend: ConnectorAdapter()} + for b in ("wasm-edge", "p2p-mesh", "volunteer-boinc", "blockchain-rlc"): ad[b] = DescriptorAdapter(b) return ad diff --git a/tools/sourceosctl.py b/tools/sourceosctl.py index 2a18984..8861d74 100644 --- a/tools/sourceosctl.py +++ b/tools/sourceosctl.py @@ -51,11 +51,12 @@ def _sha(s: str) -> str: def run_workload(*, name, command, effect, sensitivity, scalable, gpu, image, subject, heartbeats_dir, key, apply=False, dry=False, admission=None, cost=1.0, - inception=False) -> dict: + inception=False, parallelism=1, nodes=1) -> dict: """Core of `run` — testable without the CLI. Returns the full spine trace (or a placement).""" reg = mt.MeshRegistry.from_dir(heartbeats_dir) workload = {"name": name, "command": command, "effect": effect, "sensitivity": sensitivity, - "scalable": scalable, "needs_gpu": gpu, "image": image} + "scalable": scalable, "needs_gpu": gpu, "image": image, + "parallelism": parallelism, "nodes": nodes} if inception: # mount the agent-machine's persistent TopoLVM inception mount (k8s backend) import devspace workload["inception_pvc"] = devspace.INCEPTION_PVC @@ -77,12 +78,15 @@ def cmd_run(args) -> int: if _dev_mode(): print("! DEV MODE: no SOURCEOS_SIGNING_KEY set — synthesizing a dev attestation + HMAC key. " "Not for production.", file=sys.stderr) - admission = None if args.dry else adm.AdmissionController(ledger_path=LEDGER) + # `place` reuses cmd_run but its parser omits the run-only flags — read them defensively. + dry = getattr(args, "dry", False) + admission = None if dry else adm.AdmissionController(ledger_path=LEDGER) out = run_workload(name=args.name, command=args.command, effect=args.effect, sensitivity=args.sensitivity, scalable=not args.no_scale, gpu=args.gpu, image=args.image, subject=args.subject, heartbeats_dir=HEARTBEATS, - key=_key(), apply=args.apply, dry=args.dry, admission=admission, cost=args.cost, - inception=args.inception) + key=_key(), apply=getattr(args, "apply", False), dry=dry, admission=admission, + cost=getattr(args, "cost", 1.0), inception=getattr(args, "inception", False), + parallelism=getattr(args, "parallelism", 1), nodes=getattr(args, "nodes", 1)) if out["status"] == "denied": a = out["admission"] print(f"DENIED (fail-closed): {a['reason']} [usage {a['usage']} vs quota {a['quota']}]") @@ -145,6 +149,8 @@ def build_parser() -> argparse.ArgumentParser: r.add_argument("--effect", default="compute", choices=["read", "write", "compute", "exec", "egress"]) r.add_argument("--sensitivity", default="normal", choices=["normal", "sensitive"]) r.add_argument("--gpu", action="store_true", help="workload needs a GPU") + r.add_argument("--parallelism", type=int, default=1, help="N parallel tasks/ranks (k8s Indexed Job / SLURM --ntasks)") + r.add_argument("--nodes", type=int, default=1, help="node count for the parallel job (SLURM --nodes)") r.add_argument("--no-scale", action="store_true", help="keep it small (non-scalable)") r.add_argument("--subject", default="spiffe://sourceos/agent/dev", help="the requesting subject SPIFFE id") r.add_argument("--cost", type=float, default=1.0, help="cost units to charge against the subject's budget") diff --git a/tools/test_compute_plane.py b/tools/test_compute_plane.py index 387610a..a1e7ccd 100644 --- a/tools/test_compute_plane.py +++ b/tools/test_compute_plane.py @@ -81,6 +81,16 @@ def test_receipt_seal_is_deterministic_and_covers_the_decision(): assert reseal == d["receipt_digest"] +def test_connector_backend_is_external_barred_for_sensitive_but_open_to_normal(): + # a vendor connector is untrusted: sensitive work refuses it (blocks); normal work may use it. + blocked = cp.place({"sensitivity": "sensitive", "scalable": True}, + {"allowed_backends": ["connector"]}, {"connector": 100}) + assert blocked["backend"] is None + ok = cp.place({"sensitivity": "normal", "scalable": True}, + {"allowed_backends": ["connector"]}, {"connector": 100}) + assert ok["backend"] == "connector" + + def test_backends_view_exposes_the_whole_mesh_with_availability(): view = cp.backends_view({"volunteer-boinc": 200}) ids = {b["id"] for b in view["backends"]} diff --git a/tools/test_executor.py b/tools/test_executor.py index f1a31be..a6f2eac 100644 --- a/tools/test_executor.py +++ b/tools/test_executor.py @@ -91,12 +91,45 @@ def test_k8s_mounts_the_inception_pvc_when_requested(): def test_descriptor_adapter_emits_a_backend_specific_descriptor(): - d = _decision("hpc-slurm") - res = ex.execute({"command": "srun train", "effect": "compute"}, d, _grant(d, "compute"), + d = _decision("wasm-edge") + res = ex.execute({"command": "run", "effect": "compute"}, d, _grant(d, "compute"), session_id="sess_exec1", verifier=VERIFIER) - assert res["dispatch"]["kind"] == "hpc-slurm" - assert res["dispatch"]["descriptor"]["backend"] == "hpc-slurm" - assert res["dispatch"]["descriptor"]["executor_ref"] == "node://hpc-slurm" + assert res["dispatch"]["kind"] == "wasm-edge" + assert res["dispatch"]["descriptor"]["backend"] == "wasm-edge" + assert res["dispatch"]["descriptor"]["executor_ref"] == "node://wasm-edge" + + +def test_k8s_parallel_workload_emits_an_indexed_job(): + d = _decision("k8s") + res = ex.execute({"command": "python rank.py", "effect": "compute", "parallelism": 4}, + d, _grant(d, "compute"), session_id="sess_exec1", verifier=VERIFIER, apply=False) + spec = res["dispatch"]["manifest"]["spec"] + assert spec["parallelism"] == 4 and spec["completions"] == 4 + assert spec["completionMode"] == "Indexed" # each task gets JOB_COMPLETION_INDEX = its rank + + +def test_slurm_adapter_emits_a_real_sbatch_script_with_ntasks_and_srun(): + d = _decision("hpc-slurm") + res = ex.execute({"name": "train", "command": "python train.py", "effect": "compute", + "parallelism": 8, "nodes": 2, "needs_gpu": True}, + d, _grant(d, "compute"), session_id="sess_exec1", verifier=VERIFIER) + script = res["dispatch"]["script"] + assert "#SBATCH --ntasks=8" in script and "#SBATCH --nodes=2" in script + assert "#SBATCH --gres=gpu:1" in script + assert "srun python train.py" in script # srun launches the MPI ranks + assert "grant:" in script # bound to the grant + + +def test_connector_adapter_dispatches_a_grant_bound_call(): + d = _decision("connector") + grant = _grant(d, "egress") + res = ex.execute({"effect": "egress", "connector": "openai-files", "operation": "files.create", + "artifact_ref": "commons:data/corpus@1"}, + d, grant, session_id="sess_exec1", verifier=VERIFIER) + call = res["dispatch"]["call"] + assert call["connector"] == "openai-files" and call["operation"] == "files.create" + assert call["artifact_ref"] == "commons:data/corpus@1" and call["effect"] == "egress" + assert call["grant_id"] == grant["grant_id"] # grant-bound # ── fail-closed on the Grant ───────────────────────────────────────────────────────── diff --git a/tools/test_sourceosctl.py b/tools/test_sourceosctl.py index b84a093..ecd732e 100644 --- a/tools/test_sourceosctl.py +++ b/tools/test_sourceosctl.py @@ -54,6 +54,12 @@ def test_readonly_subcommands_exit_clean(): assert ctl.cmd_commons(None) == 0 +def test_place_subcommand_runs_without_run_only_flags(): + # `place` omits --cost/--inception/--parallelism; cmd_run must not AttributeError on them. + args = ctl.build_parser().parse_args(["place", "--gpu", "--sensitivity", "normal"]) + assert args.func(args) == 0 + + if __name__ == "__main__": import sys fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]