feat(executor): parallel/MPI jobs + real SLURM sbatch + connector-as-dispatch (IBM PE codified)

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 <SOURCEOS_SLURM_LOGIN> 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.

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},
 }
 
 

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 <login> 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 <login> 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
 

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")

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"]}

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 ─────────────────────────────────────────────────────────

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_")]