fix(sandbox): support Clearwing workspace transfer on Kubernetes

clearwing/sandbox/hunter_sandbox.py

@@ -346,6 +346,13 @@ def spawn(
 
         Returns a SandboxInstance ready for exec/write/read.
         """
+        if self.backend.name == "kubernetes":
+            # Kubernetes sandboxes do not share the caller's filesystem.
+            # Stream the repository after pod startup and use the image's
+            # pod-local /scratch directory instead of requesting host mounts.
+            writable_workspace = True
+            scratch_mount = False
+
         if self._environment_ref is None:
             self.prepare_environment()
 
@@ -393,30 +400,36 @@ def spawn(
 
         sb = self.backend.create(environment_ref, cfg)
         _t = time.monotonic()
-        sb.start()
-        logger.debug(
-            "Sandbox instance started environment=%s in %.2fs",
-            environment_ref,
-            time.monotonic() - _t,
-        )
+        try:
+            sb.start()
+            logger.debug(
+                "Sandbox instance started environment=%s in %.2fs",
+                environment_ref,
+                time.monotonic() - _t,
+            )
 
-        if writable_workspace:
-            sb.copy_tree_into(self.repo_path, "/workspace")
-            try:
-                baseline = sb.exec(
-                    "cd /workspace && "
-                    "rm -rf /workspace/.git && "
-                    "find . -name .git -type f -delete && "
-                    "git init -q && git add -A && "
-                    "git -c user.name=clearwing -c user.email=clearwing@localhost "
-                    "commit -m initial -q && git rev-parse HEAD",
-                    timeout=120,
-                )
-                if baseline.exit_code != 0:
-                    raise RuntimeError(baseline.stderr or "git baseline creation failed")
-                sb.workspace_baseline_commit = baseline.stdout.strip()
-            except Exception:
-                logger.warning("git init in writable workspace failed", exc_info=True)
+            if writable_workspace:
+                sb.copy_tree_into(self.repo_path, "/workspace")
+                try:
+                    baseline = sb.exec(
+                        "cd /workspace && "
+                        "rm -rf /workspace/.git && "
+                        "find . -name .git -type f -delete && "
+                        "git init -q && git add -A && "
+                        "git -c user.name=clearwing -c user.email=clearwing@localhost "
+                        "commit -m initial -q && git rev-parse HEAD",
+                        timeout=120,
+                    )
+                    if baseline.exit_code != 0:
+                        raise RuntimeError(baseline.stderr or "git baseline creation failed")
+                    sb.workspace_baseline_commit = baseline.stdout.strip()
+                except Exception:
+                    logger.warning("git init in writable workspace failed", exc_info=True)
+        except BaseException:
+            sb.stop()
+            if scratch_host_dir:
+                shutil.rmtree(scratch_host_dir, ignore_errors=True)
+            raise
 
         # Stash scratch host dir + variant on the container for cleanup / introspection
         sb.scratch_host_dir = scratch_host_dir

clearwing/sandbox/kubernetes.py

@@ -9,6 +9,7 @@
 
 import io
 import json
+import logging
 import os
 import shlex
 import subprocess
@@ -26,7 +27,8 @@
 from .container import ExecResult
 
 _STARTUP_TIMEOUT_SECONDS = 120
-_DEADLINE_FLOOR_SECONDS = 900
+
+logger = logging.getLogger(__name__)
 
 
 def kubernetes_namespace() -> str:
@@ -174,20 +176,20 @@ def start(self) -> str:
             allow_privilege_escalation=False,
             capabilities=client.V1Capabilities(drop=["ALL"], add=["SYS_PTRACE"]),
         )
+        api = self._api_factory()
+        owner_references = self._owner_references(api=api, client=client)
         pod = client.V1Pod(
             metadata=client.V1ObjectMeta(
                 name=pod_name,
                 namespace=self.namespace,
                 labels={"app.kubernetes.io/managed-by": "clearwing", "purpose": "sandbox"},
+                owner_references=owner_references,
             ),
             spec=client.V1PodSpec(
                 automount_service_account_token=False,
                 enable_service_links=False,
                 host_network=False,
                 restart_policy="Never",
-                active_deadline_seconds=max(
-                    _DEADLINE_FLOOR_SECONDS, self._config.timeout_seconds * 3
-                ),
                 containers=[
                     client.V1Container(
                         name="sandbox",
@@ -206,16 +208,42 @@ def start(self) -> str:
                 ],
             ),
         )
-        api = self._api_factory()
         api.create_namespaced_pod(self.namespace, pod)
         self._pod_name = pod_name
+        from .registry import ContainerRegistry
+
+        ContainerRegistry.get().register(self)
         try:
             self._wait_until_running(api)
         except BaseException:
             self.stop()
             raise
         return pod_name
 
+    def _owner_references(self, *, api: Any, client: Any) -> list[Any] | None:
+        """Make Kubernetes garbage-collect a sandbox if its caller pod disappears."""
+        caller_name = os.environ.get("HOSTNAME", "").strip()
+        if not caller_name:
+            return None
+        try:
+            caller = api.read_namespaced_pod(caller_name, self.namespace)
+            uid = caller.metadata.uid
+            if not isinstance(uid, str) or not uid:
+                return None
+            return [
+                client.V1OwnerReference(
+                    api_version="v1",
+                    kind="Pod",
+                    name=caller_name,
+                    uid=uid,
+                    block_owner_deletion=False,
+                    controller=False,
+                )
+            ]
+        except Exception:
+            logger.debug("Could not attach sandbox to caller pod", exc_info=True)
+            return None
+
     def _wait_until_running(self, api: Any) -> None:
         deadline = time.monotonic() + _STARTUP_TIMEOUT_SECONDS
         while time.monotonic() < deadline:
@@ -233,6 +261,15 @@ def _stream(self, command: list[str], *, stdin: bytes | None = None):
             raise RuntimeError("sandbox pod has not been started")
         from kubernetes.stream import stream
 
+        if stdin is not None:
+            # The Kubernetes Python WSClient has no half-close operation for
+            # stdin. Make the remote process consume an exact byte count so it
+            # can finish without waiting for an EOF that the client cannot send.
+            command = [
+                "/bin/sh",
+                "-c",
+                f"head -c {len(stdin)} | {shlex.join(command)}",
+            ]
         response = stream(
             self._api_factory().connect_get_namespaced_pod_exec,
             self._pod_name,
@@ -248,7 +285,6 @@ def _stream(self, command: list[str], *, stdin: bytes | None = None):
         if stdin is not None:
             for offset in range(0, len(stdin), 1024 * 1024):
                 response.write_stdin(stdin[offset : offset + 1024 * 1024])
-            response.close_stdin()
         return response
 
     def exec(
@@ -340,16 +376,29 @@ def copy_tree_into(self, host_path: str, container_path: str = "/workspace") ->
     def stop(self) -> None:
         if self._pod_name is None:
             return
-        pod_name, self._pod_name = self._pod_name, None
+        pod_name = self._pod_name
         try:
             self._api_factory().delete_namespaced_pod(
                 pod_name, self.namespace, grace_period_seconds=5
             )
-        except Exception:
-            pass
+        except Exception as error:
+            if getattr(error, "status", None) != 404:
+                logger.warning("Failed to delete Kubernetes sandbox pod %s", pod_name, exc_info=True)
+                return
+        from .registry import ContainerRegistry
+
+        ContainerRegistry.get().unregister(self)
+        self._pod_name = None
 
     def __enter__(self) -> KubernetesSandbox:
         return self
 
     def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.stop()
+
+    def __del__(self) -> None:
+        if self._pod_name is not None:
+            try:
+                self.stop()
+            except Exception:
+                logger.debug("Kubernetes sandbox garbage-collection cleanup failed", exc_info=True)

docs/sandbox-backends.md

@@ -111,6 +111,20 @@ The returned `SandboxInstance` provides `start`, `exec`, `write_file`,
 lifecycle properties. Downstream SourceHunt code depends on this protocol,
 not on the Docker SDK implementation.
 
+### Kubernetes lifecycle
+
+Kubernetes sandbox pods intentionally have no pod-wide
+`activeDeadlineSeconds`: SourceHunts may legitimately outlive the default
+per-command timeout. Individual `exec` calls still enforce their requested
+timeouts, and the caller owns the overall SourceHunt deadline.
+
+Clearwing deletes each pod during normal cleanup, retries a failed deletion at
+process exit, and records the calling pod as an owner when it can be resolved.
+The owner reference is the hard-crash safety net: Kubernetes garbage-collects
+the sandbox after the calling pod is deleted even when Clearwing cannot run
+cleanup. Deployments should apply a TTL to completed caller Jobs so that this
+fallback is timely.
+
 ## JSON-RPC protocol
 
 The socket adapter uses JSON-RPC 2.0. Each request or response is one UTF-8

tests/test_kubernetes_sandbox.py

@@ -42,6 +42,45 @@ def test_host_mounts_are_rejected():
         backend.create("sandbox:1", config)
 
 
+def test_hunter_sandbox_streams_workspace_and_uses_pod_local_scratch(tmp_path):
+    from clearwing.sandbox.hunter_sandbox import HunterSandbox
+
+    backend = MagicMock(name="backend")
+    backend.name = "kubernetes"
+    backend.available_cpus.return_value = (None, "Kubernetes scheduler")
+    backend.ensure_environment.return_value = SimpleNamespace(reference="sandbox:1", cached=True)
+    instance = MagicMock()
+    instance.scratch_host_dir = None
+    backend.create.return_value = instance
+    manager = HunterSandbox(str(tmp_path), backend=backend)
+
+    manager.spawn()
+
+    config = backend.create.call_args.args[1]
+    assert config.mounts == []
+    instance.copy_tree_into.assert_called_once_with(str(tmp_path), "/workspace")
+
+
+def test_hunter_sandbox_stops_pod_when_workspace_transfer_fails(tmp_path):
+    from clearwing.sandbox.hunter_sandbox import HunterSandbox
+
+    backend = MagicMock(name="backend")
+    backend.name = "kubernetes"
+    backend.available_cpus.return_value = (None, "Kubernetes scheduler")
+    backend.ensure_environment.return_value = SimpleNamespace(reference="sandbox:1", cached=True)
+    instance = MagicMock()
+    instance.scratch_host_dir = None
+    instance.copy_tree_into.side_effect = RuntimeError("transfer failed")
+    backend.create.return_value = instance
+    manager = HunterSandbox(str(tmp_path), backend=backend)
+
+    with pytest.raises(RuntimeError, match="transfer failed"):
+        manager.spawn()
+
+    instance.stop.assert_called_once_with()
+    assert manager._spawned == []
+
+
 def test_pod_lifecycle(monkeypatch):
     client = pytest.importorskip("kubernetes.client")
     api = MagicMock()
@@ -57,15 +96,85 @@ def test_pod_lifecycle(monkeypatch):
     assert identity.startswith("clearwing-sandbox-")
     assert sandbox.is_running
     api.create_namespaced_pod.assert_called_once()
+    pod = api.create_namespaced_pod.call_args.args[1]
+    assert pod["spec"].active_deadline_seconds is None
 
     sandbox.stop()
     api.delete_namespaced_pod.assert_called_once()
     assert sandbox.container_id is None
 
 
+def test_pod_is_owned_by_caller_pod(monkeypatch):
+    client = pytest.importorskip("kubernetes.client")
+    api = MagicMock()
+    api.read_namespaced_pod.return_value = SimpleNamespace(
+        metadata=SimpleNamespace(uid="caller-uid")
+    )
+    api.read_namespaced_pod_status.return_value = SimpleNamespace(
+        status=SimpleNamespace(phase="Running")
+    )
+    sandbox = KubernetesSandbox(
+        "sandbox:1", SandboxRunConfig(), namespace="jobs", api_factory=lambda: api
+    )
+    monkeypatch.setenv("HOSTNAME", "dagster-run-123")
+    monkeypatch.setattr(client, "V1Pod", MagicMock(side_effect=lambda **kw: kw))
+
+    sandbox.start()
+
+    pod = api.create_namespaced_pod.call_args.args[1]
+    owner = pod["metadata"].owner_references[0]
+    assert owner.name == "dagster-run-123"
+    assert owner.uid == "caller-uid"
+    sandbox.stop()
+
+
+def test_failed_delete_remains_registered_for_retry(monkeypatch):
+    pytest.importorskip("kubernetes.client")
+    from clearwing.sandbox.registry import ContainerRegistry
+
+    api = MagicMock()
+    api.read_namespaced_pod_status.return_value = SimpleNamespace(
+        status=SimpleNamespace(phase="Running")
+    )
+    sandbox = KubernetesSandbox(
+        "sandbox:1", SandboxRunConfig(), namespace="jobs", api_factory=lambda: api
+    )
+    monkeypatch.delenv("HOSTNAME", raising=False)
+    sandbox.start()
+    assert ContainerRegistry.get().active_count >= 1
+    api.delete_namespaced_pod.side_effect = RuntimeError("temporary API failure")
+
+    sandbox.stop()
+
+    assert sandbox.container_id is not None
+    api.delete_namespaced_pod.side_effect = None
+    sandbox.stop()
+    assert sandbox.container_id is None
+
+
 def test_exit_code_parsing():
     response = MagicMock()
     response.read_channel.return_value = (
         '{"status":"Failure","details":{"causes":[{"reason":"ExitCode","message":"137"}]}}'
     )
     assert KubernetesSandbox._parse_exit_code(response) == 137
+
+
+def test_stream_with_stdin_uses_exact_length_instead_of_half_close(monkeypatch):
+    import kubernetes.stream
+
+    response = MagicMock(spec=["write_stdin"])
+    stream = MagicMock(return_value=response)
+    monkeypatch.setattr(kubernetes.stream, "stream", stream)
+    sandbox = KubernetesSandbox("sandbox:1", SandboxRunConfig(), namespace="jobs")
+    sandbox._pod_name = "clearwing-sandbox-test"
+
+    result = sandbox._stream(["tar", "-xf", "-", "-C", "/workspace"], stdin=b"payload")
+
+    assert result is response
+    assert stream.call_args.kwargs["command"] == [
+        "/bin/sh",
+        "-c",
+        "head -c 7 | tar -xf - -C /workspace",
+    ]
+    response.write_stdin.assert_called_once_with(b"payload")