feat(deps): base-image build recipe + wired-image drift guard

apps/api/pytest.ini

@@ -35,5 +35,6 @@ python_files = test_suite_v060.py test_suite_v062.py test_suite_v0610.py test_se
     test_agent_redeploy.py
     test_package_revocation.py
     test_deps_allowlist.py
+    test_base_deps_recipe.py
 markers =
     no_api_key: test does not require WAYFORTH_TEST_API_KEY (e.g. probes unauthenticated paths)

apps/api/scripts/build_base_image.py

@@ -0,0 +1,70 @@
+"""scripts/build_base_image.py — reproducibly bake the agent base image from BASE_DEPS.
+
+The base image (AGENT_BASE_IMAGE) carries the full resolved dependency closure so that,
+under gateway-only egress, agent runs need no run-time pip. This script bakes it FROM
+services.agent_deps.BASE_DEPS — the same source the ship-gate verifies against — so the
+wired image can't drift from the recipe (the ⚠ gap the flip-readiness check found: the
+old image was baked by an uncommitted manual step).
+
+Gated — does nothing unless BUILD_BASE_IMAGE=1 and E2B_API_KEY is set:
+
+    BUILD_BASE_IMAGE=1 E2B_API_KEY=... [DEPS_MIRROR_URL=...] \
+        /app/.venv/bin/python -m scripts.build_base_image
+
+It builds in a mirror-only-egress sandbox, installs the hashed wheels-only closure,
+ASSERTS the built `pip freeze` equals BASE_DEPS before snapshotting, then prints the ref.
+It does NOT set AGENT_BASE_IMAGE and does NOT flip anything — wiring is a separate step.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+from services.agent_deps import (
+    base_deps_lock, base_deps_match, diff_against_base_deps, pip_install_command,
+)
+
+MIRROR_URL = os.environ.get("DEPS_MIRROR_URL", "https://pypi.org/simple")
+MIRROR_HOSTS = os.environ.get("DEPS_MIRROR_HOSTS", "pypi.org,files.pythonhosted.org").split(",")
+SNAPSHOT_NAME = os.environ.get("BASE_IMAGE_NAME", "wayforth-agent-base-v1")
+
+
+def _net(allow):
+    from e2b import SandboxNetworkOpts
+    return SandboxNetworkOpts(deny_out=["0.0.0.0/0"], allow_out=list(allow))
+
+
+def main() -> int:
+    if os.environ.get("BUILD_BASE_IMAGE") != "1" or not os.environ.get("E2B_API_KEY"):
+        print("build_base_image: skipped (set BUILD_BASE_IMAGE=1 and E2B_API_KEY).")
+        return 0
+
+    from e2b import Sandbox
+
+    b = Sandbox.create(timeout=300, network=_net(MIRROR_HOSTS))
+    try:
+        b.files.write("/home/user/requirements.lock", base_deps_lock())
+        r = b.commands.run(pip_install_command(MIRROR_URL) + " ; echo EXIT=$?", timeout=240)
+        if "EXIT=0" not in (r.stdout or ""):
+            print("FAIL: base install failed\n", (r.stderr or "")[-500:])
+            return 1
+        # the recipe guarantee: the built closure must EXACTLY equal BASE_DEPS before we bake
+        fr = b.commands.run("python3 -m pip list --format=freeze")
+        if not base_deps_match(fr.stdout or ""):
+            print("FAIL: built closure != BASE_DEPS — not snapshotting.")
+            print("  diff:", diff_against_base_deps(fr.stdout or ""))
+            return 1
+        snap = b.create_snapshot(name=SNAPSHOT_NAME)
+        ref = getattr(snap, "snapshot_id", None) or getattr(snap, "template_id", None) or SNAPSHOT_NAME
+        print("BASE IMAGE BUILT — closure matches BASE_DEPS exactly.")
+        print(f"  name: {SNAPSHOT_NAME}")
+        print(f"  ref:  {ref}")
+        print("  next (manual, separate steps): set AGENT_BASE_IMAGE to this ref →")
+        print("        run deps_live_proof (asserts wired image == BASE_DEPS) → flip egress flag.")
+        return 0
+    finally:
+        b.kill()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

apps/api/scripts/deps_live_proof.py

@@ -25,7 +25,8 @@
 import sys
 
 from services.agent_deps import (
-    BASE_DEPS, _norm, build_requirements_lock, load_lockfile, pip_install_command,
+    base_deps_lock, base_deps_match, build_requirements_lock, diff_against_base_deps,
+    pip_install_command,
 )
 
 MIRROR_URL = os.environ.get("DEPS_MIRROR_URL", "https://pypi.org/simple")
@@ -49,9 +50,7 @@ def main() -> int:
 
     from e2b import Sandbox
 
-    lock = load_lockfile()
-    base = [(_norm(n), v, lock[_norm(n)][v]) for n, v in BASE_DEPS]
-    good_lock = build_requirements_lock(base)
+    good_lock = base_deps_lock()  # the BASE_DEPS closure — shared with build_base_image.py
     bad_lock = build_requirements_lock([("httpx", "0.28.1", ["sha256:" + "0" * 64])])
     pip = pip_install_command(MIRROR_URL)
     failures = []
@@ -110,6 +109,25 @@ def main() -> int:
             except Exception:
                 pass
 
+    # WIRED-IMAGE ANTI-DRIFT — if AGENT_BASE_IMAGE is set, the image agents will actually
+    # boot must equal BASE_DEPS exactly. This is what ties "proof passed" to "wired image
+    # correct": you set AGENT_BASE_IMAGE, run this, and a stale/hand-built image fails here.
+    wired = os.environ.get("AGENT_BASE_IMAGE", "").strip()
+    if wired:
+        wb = Sandbox.create(wired, timeout=120, network=_net([GATEWAY_HOST]))
+        try:
+            fr = _run(wb, "python3 -m pip list --format=freeze")
+            if not base_deps_match(fr.stdout or ""):
+                failures.append(
+                    f"WIRED base image (AGENT_BASE_IMAGE={wired}) != BASE_DEPS: "
+                    f"{diff_against_base_deps(fr.stdout or '')}")
+            else:
+                print(f"WIRED PASS — AGENT_BASE_IMAGE={wired} closure matches BASE_DEPS exactly")
+        finally:
+            wb.kill()
+    else:
+        print("WIRED check skipped — AGENT_BASE_IMAGE not set (set it before flipping).")
+
     if failures:
         print("DEPS LIVE PROOF: FAIL")
         for f in failures:

apps/api/services/agent_deps.py

@@ -171,6 +171,49 @@ def build_requirements_lock(install_set) -> str:
     return "\n".join(lines) + "\n"
 
 
+# ── base-image recipe + drift detection (shared by the build script + the ship-gate) ──
+# These exist so "the wired base image" can never silently drift from BASE_DEPS: the build
+# script bakes FROM base_deps_lock(), and deps_live_proof asserts the wired image's
+# pip freeze matches base_deps_pinned(). Single source of truth = BASE_DEPS.
+
+_BASE_IGNORE = {"pip", "setuptools", "wheel"}  # base interpreter pkgs, not part of the closure
+
+
+def base_deps_lock() -> str:
+    """The hashed requirements.lock for the full BASE_DEPS closure (what the base image
+    bakes). Pulls the hashes from the allowlist lockfile."""
+    lock = load_lockfile()
+    base = [(_norm(n), v, lock[_norm(n)][v]) for n, v in BASE_DEPS]
+    return build_requirements_lock(base)
+
+
+def base_deps_pinned() -> dict:
+    """{normalized_name: version} for BASE_DEPS — the expected baked closure."""
+    return {_norm(n): v for n, v in BASE_DEPS}
+
+
+def diff_against_base_deps(freeze_text: str) -> dict:
+    """Compare a `pip list --format=freeze` (or `pip freeze`) dump to BASE_DEPS.
+    Returns {missing, mismatch, extra}; all empty ⇒ the image is exactly BASE_DEPS."""
+    got: dict[str, str] = {}
+    for line in freeze_text.splitlines():
+        line = line.strip()
+        if "==" in line:
+            name, _, ver = line.partition("==")
+            got[_norm(name)] = ver.strip()
+    expected = base_deps_pinned()
+    missing = sorted(n for n in expected if n not in got)
+    mismatch = sorted((n, expected[n], got[n]) for n in expected if n in got and got[n] != expected[n])
+    extra = sorted(n for n in got if n not in expected and n not in _BASE_IGNORE)
+    return {"missing": missing, "mismatch": mismatch, "extra": extra}
+
+
+def base_deps_match(freeze_text: str) -> bool:
+    """True iff the freeze dump equals BASE_DEPS exactly (no missing/mismatch/extra)."""
+    d = diff_against_base_deps(freeze_text)
+    return not (d["missing"] or d["mismatch"] or d["extra"])
+
+
 def pip_install_command(mirror_url: str, reqs_path: str = _REQS_PATH) -> str:
     """The wheels-only, hashed, mirror-pinned, no-deps install — the install-RCE control."""
     return (

apps/api/tests/test_base_deps_recipe.py

@@ -0,0 +1,70 @@
+"""test_base_deps_recipe.py — the base-image recipe + drift detector (Step: flip-readiness).
+
+Pure functions shared by scripts/build_base_image.py (bakes FROM BASE_DEPS) and
+scripts/deps_live_proof.py (asserts the WIRED image == BASE_DEPS), so "proof passed" and
+"wired image correct" can't drift.
+"""
+from services.agent_deps import (
+    BASE_DEPS, base_deps_lock, base_deps_match, base_deps_pinned, diff_against_base_deps,
+)
+
+# The ACTUAL pip freeze captured from the live wayforth-agent-base-v1 template (read-only
+# inspection). The drift assertion must PASS on this — the wired image is currently correct.
+CURRENT_TEMPLATE_FREEZE = """\
+anyio==4.14.1
+certifi==2026.6.17
+h11==0.16.0
+httpcore==1.0.9
+httpx==0.28.1
+idna==3.18
+pip==23.2.1
+setuptools==65.5.1
+typing_extensions==4.15.0
+wayforth-sdk==0.9.0
+wheel==0.42.0
+"""
+
+
+def test_pinned_has_the_critical_closure():
+    p = base_deps_pinned()
+    # the lazy-import deps #61 caught + the SDK/http client
+    for name in ("httpcore", "typing-extensions", "wayforth-sdk", "httpx"):
+        assert name in p, f"{name} missing from BASE_DEPS"
+    assert len(p) == len(BASE_DEPS)
+
+
+def test_current_template_matches_base_deps():
+    # underscore typing_extensions normalizes to typing-extensions; pip/setuptools/wheel ignored
+    assert diff_against_base_deps(CURRENT_TEMPLATE_FREEZE) == {"missing": [], "mismatch": [], "extra": []}
+    assert base_deps_match(CURRENT_TEMPLATE_FREEZE) is True
+
+
+def test_detects_missing_dep():
+    freeze = "\n".join(l for l in CURRENT_TEMPLATE_FREEZE.splitlines() if "httpcore" not in l)
+    d = diff_against_base_deps(freeze)
+    assert d["missing"] == ["httpcore"]
+    assert base_deps_match(freeze) is False
+
+
+def test_detects_version_mismatch():
+    freeze = CURRENT_TEMPLATE_FREEZE.replace("wayforth-sdk==0.9.0", "wayforth-sdk==0.8.0")
+    d = diff_against_base_deps(freeze)
+    assert d["mismatch"] == [("wayforth-sdk", "0.9.0", "0.8.0")]
+    assert base_deps_match(freeze) is False
+
+
+def test_detects_unexpected_extra():
+    freeze = CURRENT_TEMPLATE_FREEZE + "requests==2.32.0\n"   # not in the closure
+    d = diff_against_base_deps(freeze)
+    assert d["extra"] == ["requests"]
+    assert base_deps_match(freeze) is False
+
+
+def test_base_deps_lock_is_hashed_and_covers_closure():
+    lock = base_deps_lock()
+    assert "httpcore==1.0.9" in lock and "--hash=sha256:" in lock
+    assert "typing-extensions==4.15.0" in lock
+    # one line per BASE_DEPS entry, every line hashed
+    lines = [l for l in lock.splitlines() if l.strip()]
+    assert len(lines) == len(BASE_DEPS)
+    assert all("--hash=sha256:" in l for l in lines)