feat(code-editing): Step 2 — run-sandbox egress lock (gateway-only), flag-gated

apps/api/pytest.ini

@@ -30,5 +30,6 @@ python_files = test_suite_v060.py test_suite_v062.py test_suite_v0610.py test_se
     test_params_eval.py
     test_templates_params.py
     test_agent_deps.py
+    test_sandbox_egress.py
 markers =
     no_api_key: test does not require WAYFORTH_TEST_API_KEY (e.g. probes unauthenticated paths)

apps/api/scripts/deps_live_proof.py

@@ -89,11 +89,15 @@ def main() -> int:
     if sid:
         rn = Sandbox.create(sid, timeout=120, network=_net([GATEWAY_HOST]))
         try:
-            imp = _run(rn, 'python3 -c "import httpx;print(httpx.__version__)" && pip show wayforth-sdk | grep -i ^version')
-            if "EXIT=0" not in imp.stdout:
-                failures.append(f"#3 base deps not importable in run sandbox: {imp.stdout!r}")
+            # A REAL httpx request (not just `import httpx`) — exercises the full
+            # closure (httpcore et al.), the gap a bare import check misses.
+            imp = _run(rn, 'python3 -c "import httpx; '
+                       f'r=httpx.get(\\"https://{GATEWAY_HOST}/status\\",timeout=10); '
+                       'print(\\"OK\\", httpx.__version__, r.status_code)"')
+            if "EXIT=0" not in imp.stdout or "OK" not in imp.stdout:
+                failures.append(f"#3 existing-agent workload failed in run sandbox: {imp.stdout!r}")
             else:
-                print(f"#3 PASS — base deps importable in run sandbox: {imp.stdout.splitlines()[:2]}")
+                print(f"#3 PASS — real httpx request works on base image: {imp.stdout.splitlines()[:1]}")
             # §0: run sandbox cannot reach the mirror/PyPI
             pp = _run(rn, f'curl -sS -o /dev/null -w "%{{http_code}}" --max-time 8 {MIRROR_URL}')
             if "ec=35" not in pp.stdout and "000" not in pp.stdout:

apps/api/services/agent_deps.py

@@ -34,14 +34,19 @@
 # Base closure baked into EVERY agent image. Gateway-only run egress (Step 2) makes
 # run-time pip impossible, so the SDK + http client (and httpx's pinned closure) live
 # in the image. All present in the lockfile.
+# The FULL resolved closure (pip download wayforth-sdk httpx, with deps) — not a
+# hand-list. The survival proof (a real httpx request, not just `import httpx`) caught
+# that hand-listing missed httpcore + typing-extensions; httpx imports httpcore lazily
+# only on the first request, so an import check alone passed while a real run failed.
 BASE_DEPS = [
-    ("wayforth-sdk", "0.9.0"),
-    ("httpx", "0.28.1"),
     ("anyio", "4.14.1"),
-    ("sniffio", "1.3.1"),
-    ("h11", "0.16.0"),
     ("certifi", "2026.6.17"),
+    ("h11", "0.16.0"),
+    ("httpcore", "1.0.9"),
+    ("httpx", "0.28.1"),
     ("idna", "3.18"),
+    ("typing-extensions", "4.15.0"),
+    ("wayforth-sdk", "0.9.0"),
 ]
 
 _REQS_PATH = "/home/user/requirements.lock"

apps/api/services/agent_deps_lock.json

@@ -24,6 +24,11 @@
       "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86"
     ]
   },
+  "httpcore": {
+    "1.0.9": [
+      "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55"
+    ]
+  },
   "httpx": {
     "0.28.1": [
       "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad"
@@ -54,6 +59,11 @@
       "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2"
     ]
   },
+  "typing-extensions": {
+    "4.15.0": [
+      "sha256:f0fa19c6845758ab08074a0cfa8b7aecb71c999ca73d62883bc25cc018c4e548"
+    ]
+  },
   "urllib3": {
     "2.7.0": [
       "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897"
@@ -64,4 +74,4 @@
       "sha256:234d07cc3646fc7887873a991e626ae2432a371724a87632abf4081ac15bd95a"
     ]
   }
-}
+}

apps/api/services/sandbox.py

@@ -38,6 +38,26 @@
     "fc00::/7",         # IPv6 ULA
 ]
 
+# ── Step 2: run-sandbox egress lock (gateway-only) ──────────────────────────────
+# Flag-gated cutover (default OFF), reversible like the run-token flip. When ON, a
+# PYTHON run gets gateway-ONLY egress (deny everything, allow only the gateway — no
+# exfiltration) and boots the pre-built BASE IMAGE (httpx/wayforth-sdk closure baked),
+# so it runs WITHOUT the run-time `pip install` that gateway-only egress would break.
+# Requires AGENT_BASE_IMAGE to be set (the built base snapshot); if unset, falls back
+# to the current deny-list path so a run never breaks. node runs are NOT locked in v1
+# (they npm-install at run time; a node deps pipeline is a follow-on).
+_GATEWAY_HOST = os.environ.get("WAYFORTH_GATEWAY_HOST", "gateway.wayforth.io")
+
+
+def _gateway_egress_enabled() -> bool:
+    return os.environ.get("AGENT_GATEWAY_EGRESS_ENABLED", "").strip().lower() in (
+        "1", "true", "yes", "on",
+    )
+
+
+def _agent_base_image() -> str:
+    return os.environ.get("AGENT_BASE_IMAGE", "").strip()
+
 
 @dataclass
 class SandboxResult:
@@ -94,22 +114,39 @@ def _run_sync(
     ) -> SandboxResult:
         from e2b import Sandbox, SandboxNetworkOpts
 
-        network = SandboxNetworkOpts(deny_out=_DENY_EGRESS)
-        t0 = time.monotonic()
-        sbx = Sandbox.create(
-            timeout=timeout_seconds,
-            envs=env,
-            network=network,
-            api_key=self._api_key or None,
+        # Egress lock applies to PYTHON only, and only when the base image is set
+        # (deps must be baked since gateway-only egress makes run-time pip impossible).
+        python_locked = (
+            _gateway_egress_enabled()
+            and runtime == "python3.12"
+            and bool(_agent_base_image())
         )
+        if _gateway_egress_enabled() and runtime == "python3.12" and not _agent_base_image():
+            logger.error("AGENT_GATEWAY_EGRESS_ENABLED on but AGENT_BASE_IMAGE unset; "
+                         "falling back to the deny-list path so the run doesn't break")
+
+        create_kwargs = dict(timeout=timeout_seconds, envs=env, api_key=self._api_key or None)
+        if python_locked:
+            create_kwargs["network"] = SandboxNetworkOpts(
+                deny_out=["0.0.0.0/0"], allow_out=[_GATEWAY_HOST])
+            create_kwargs["template"] = _agent_base_image()
+        else:
+            create_kwargs["network"] = SandboxNetworkOpts(deny_out=_DENY_EGRESS)
+
+        t0 = time.monotonic()
+        sbx = Sandbox.create(**create_kwargs)
         sandbox_id = sbx.sandbox_id
         try:
             if runtime == "python3.12":
                 sbx.files.write("/home/user/agent.py", code)
-                cmd = (
-                    "pip install wayforth-sdk httpx -q --break-system-packages "
-                    "2>/dev/null; python3 /home/user/agent.py"
-                )
+                if python_locked:
+                    # deps are baked into the base image — no run-time pip
+                    cmd = "python3 /home/user/agent.py"
+                else:
+                    cmd = (
+                        "pip install wayforth-sdk httpx -q --break-system-packages "
+                        "2>/dev/null; python3 /home/user/agent.py"
+                    )
             else:  # node20
                 sbx.files.write("/home/user/agent.ts", code)
                 # "type":"module" is required for top-level await in tsx/esbuild

apps/api/tests/test_sandbox_egress.py

@@ -0,0 +1,107 @@
+"""test_sandbox_egress.py — code-editing v1, Step 2 (run-sandbox egress lock).
+
+Flag-gated cutover: when AGENT_GATEWAY_EGRESS_ENABLED is on (and AGENT_BASE_IMAGE is
+set), a PYTHON run gets gateway-only egress + the base image + NO run-time pip. Flag
+off is byte-identical to today (deny-list + pip). node is never locked in v1. Falls
+back safely if the base image is unset.
+"""
+from __future__ import annotations
+
+import pytest
+
+import services.sandbox as sb
+
+
+@pytest.fixture
+def capture(monkeypatch):
+    cap = {"writes": {}}
+
+    class _Res:
+        stdout, stderr, exit_code = "ok", "", 0
+
+    class _Sbx:
+        sandbox_id = "sb-1"
+
+        def __init__(self):
+            self.files = self
+            self.commands = self
+
+        def write(self, path, content):
+            cap["writes"][path] = content
+
+        def run(self, cmd, timeout=None):
+            cap["cmd"] = cmd
+            return _Res()
+
+        def kill(self):
+            pass
+
+    def _create(**kwargs):
+        cap["create_kwargs"] = kwargs
+        return _Sbx()
+
+    monkeypatch.setattr("e2b.Sandbox.create", staticmethod(_create))
+    # default-off, no base image, no leftover gateway host override
+    monkeypatch.delenv("AGENT_GATEWAY_EGRESS_ENABLED", raising=False)
+    monkeypatch.delenv("AGENT_BASE_IMAGE", raising=False)
+    return cap
+
+
+def _run(runtime="python3.12"):
+    return sb.E2BSandboxProvider()._run_sync("print(1)", runtime, {}, 60)
+
+
+# ── flag OFF — byte-identical to today ──────────────────────────────────────────
+
+def test_flag_off_uses_denylist_and_pip(capture):
+    _run("python3.12")
+    net = capture["create_kwargs"]["network"]
+    assert net["deny_out"] == sb._DENY_EGRESS and "allow_out" not in net
+    assert "template" not in capture["create_kwargs"]
+    assert "pip install" in capture["cmd"]
+
+
+# ── flag ON + base image — gateway-only, base template, no pip ───────────────────
+
+def test_flag_on_python_locked(capture, monkeypatch):
+    monkeypatch.setenv("AGENT_GATEWAY_EGRESS_ENABLED", "1")
+    monkeypatch.setenv("AGENT_BASE_IMAGE", "team/wayforth-agent-base-v1:default")
+    _run("python3.12")
+    ck = capture["create_kwargs"]
+    assert ck["network"]["deny_out"] == ["0.0.0.0/0"]
+    assert ck["network"]["allow_out"] == ["gateway.wayforth.io"]
+    assert ck["template"] == "team/wayforth-agent-base-v1:default"
+    assert capture["cmd"] == "python3 /home/user/agent.py"        # NO pip
+    assert "pip install" not in capture["cmd"]
+
+
+# ── flag ON + node — NOT locked in v1 ───────────────────────────────────────────
+
+def test_flag_on_node_not_locked(capture, monkeypatch):
+    monkeypatch.setenv("AGENT_GATEWAY_EGRESS_ENABLED", "1")
+    monkeypatch.setenv("AGENT_BASE_IMAGE", "team/base:default")
+    _run("node20")
+    ck = capture["create_kwargs"]
+    assert ck["network"]["deny_out"] == sb._DENY_EGRESS          # node keeps deny-list
+    assert "template" not in ck
+    assert "npm install" in capture["cmd"]
+
+
+# ── flag ON but base image UNSET — safe fallback (never break a run) ─────────────
+
+def test_flag_on_without_base_image_falls_back(capture, monkeypatch):
+    monkeypatch.setenv("AGENT_GATEWAY_EGRESS_ENABLED", "1")
+    monkeypatch.delenv("AGENT_BASE_IMAGE", raising=False)
+    _run("python3.12")
+    ck = capture["create_kwargs"]
+    assert ck["network"]["deny_out"] == sb._DENY_EGRESS          # fell back to deny-list
+    assert "template" not in ck
+    assert "pip install" in capture["cmd"]                       # and to pip
+
+
+@pytest.mark.parametrize("val", ["0", "false", "no", "", "off"])
+def test_flag_values_treated_as_off(capture, monkeypatch, val):
+    monkeypatch.setenv("AGENT_GATEWAY_EGRESS_ENABLED", val)
+    monkeypatch.setenv("AGENT_BASE_IMAGE", "team/base:default")
+    _run("python3.12")
+    assert capture["create_kwargs"]["network"]["deny_out"] == sb._DENY_EGRESS