fix(executor): real k8s dispatch — explicit-context safety + create-not-apply

docs/EXECUTION_SPINE.md

@@ -32,8 +32,11 @@ mesh_telemetry           compute_plane        mcp_a2a_grant           mcp_a2a_gr
    attestation, effect. No valid Grant → **`DispatchRefused`**, nothing runs.
 5. **Dispatch** — a per-backend adapter runs it:
    - **local** — a real subprocess (`--apply`).
-   - **k8s** — a real, Grant-labelled `batch/v1` Job manifest (applied via `kubectl` when a cluster
-     is reachable, else emitted).
+   - **k8s** — a real, Grant-labelled `batch/v1` Job manifest, `kubectl create`-d (not `apply` — a
+     Job is one-shot/immutable and `apply` rejects `generateName`) into an **explicit** target
+     context. Applying requires `SOURCEOS_KUBE_CONTEXT` — the executor refuses to dispatch to
+     whatever kube-context happens to be current (which could be prod). Validated client-side with
+     `kubectl create --dry-run=client`.
    - **hpc-slurm · wasm-edge · p2p-mesh · volunteer-boinc · blockchain-rlc** — the substrate-specific
      descriptor to hand that scheduler over a Grant-bound channel.
 6. **Receipt** — every dispatch is hash-sealed.

tools/executor.py

@@ -86,13 +86,27 @@ def manifest(self, workload, decision, grant):
         }
 
     def dispatch(self, workload, decision, grant, *, apply):
+        import os
+        context = os.environ.get("SOURCEOS_KUBE_CONTEXT")
+        namespace = os.environ.get("SOURCEOS_KUBE_NAMESPACE", "sourceos-mesh")
         manifest = self.manifest(workload, decision, grant)
-        if apply and shutil.which("kubectl"):
-            proc = subprocess.run(["kubectl", "apply", "-f", "-"], input=json.dumps(manifest),
-                                  capture_output=True, text=True, timeout=60)
-            return {"kind": "k8s", "applied": proc.returncode == 0, "manifest": manifest,
-                    "kubectl": proc.stdout.strip() or proc.stderr.strip()}
-        return {"kind": "k8s", "applied": False, "manifest": manifest}
+        manifest["metadata"]["namespace"] = namespace
+        if not apply:
+            return {"kind": "k8s", "applied": False, "namespace": namespace, "manifest": manifest}
+        # Safety: NEVER dispatch to whatever kube-context happens to be current (that could be prod).
+        # Applying requires an explicit target context named in SOURCEOS_KUBE_CONTEXT.
+        if not context:
+            return {"kind": "k8s", "applied": False, "namespace": namespace, "manifest": manifest,
+                    "reason": "refusing to apply without SOURCEOS_KUBE_CONTEXT — won't dispatch to the current context"}
+        if not shutil.which("kubectl"):
+            return {"kind": "k8s", "applied": False, "namespace": namespace, "manifest": manifest,
+                    "reason": "kubectl not found"}
+        # `create`, not `apply`: a Job is one-shot + immutable, and `apply` rejects generateName.
+        cmd = ["kubectl", "--context", context, "create", "-n", namespace, "-f", "-"]
+        proc = subprocess.run(cmd, input=json.dumps(manifest), capture_output=True, text=True, timeout=60)
+        return {"kind": "k8s", "applied": proc.returncode == 0, "namespace": namespace,
+                "context": context, "manifest": manifest,
+                "kubectl": proc.stdout.strip() or proc.stderr.strip()}
 
 
 class DescriptorAdapter:

tools/test_executor.py

@@ -60,6 +60,27 @@ def test_k8s_adapter_emits_a_valid_grant_labelled_job():
     assert m["spec"]["template"]["spec"]["restartPolicy"] == "Never"
 
 
+def test_k8s_manifest_carries_the_target_namespace():
+    d = _decision("k8s")
+    res = ex.execute({"command": "echo x", "effect": "exec"}, d, _grant(d, "exec"),
+                     session_id="sess_exec1", verifier=VERIFIER, apply=False)
+    assert res["dispatch"]["manifest"]["metadata"]["namespace"] == "sourceos-mesh"
+
+
+def test_k8s_apply_refuses_without_an_explicit_context():
+    # applying must NOT fall back to the current kube-context (could be prod) — it needs an explicit one.
+    import os
+    saved = os.environ.pop("SOURCEOS_KUBE_CONTEXT", None)
+    try:
+        d = _decision("k8s")
+        res = ex.execute({"command": "echo x", "effect": "exec"}, d, _grant(d, "exec"),
+                         session_id="sess_exec1", verifier=VERIFIER, apply=True)
+        assert res["dispatch"]["applied"] is False and "context" in res["dispatch"]["reason"]
+    finally:
+        if saved is not None:
+            os.environ["SOURCEOS_KUBE_CONTEXT"] = saved
+
+
 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"),