feat(code-editing): Step 4 — save→redeploy orchestration (flag-gated, inert)

apps/api/main.py

@@ -1190,6 +1190,16 @@ async def lifespan(app: FastAPI):
                 FROM agent_versions av
                 WHERE av.agent_id = h.id AND av.version_no = 1 AND h.active_version_id IS NULL
             """)
+            # Migration 070: bind each run to its version (in-flight isolation anchor).
+            # Mirrored in infra/migrations/070_agent_run_version.sql. Additive/nullable;
+            # dispatch unchanged until AGENT_VERSIONED_DISPATCH_ENABLED flips (Step 4).
+            await _mconn.execute("""
+                ALTER TABLE agent_runs
+                    ADD COLUMN IF NOT EXISTS version_id UUID REFERENCES agent_versions(id)
+            """)
+            await _mconn.execute("""
+                CREATE INDEX IF NOT EXISTS idx_agent_runs_version ON agent_runs(version_id)
+            """)
             await _mconn.execute("""
                 CREATE INDEX IF NOT EXISTS idx_hosted_agents_next_run
                     ON hosted_agents(next_run_at)

apps/api/pytest.ini

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

apps/api/routers/cloud.py

@@ -46,7 +46,10 @@
 from core.db import get_db
 from core.rate_limit import limiter
 from core.tier_gates import require_tier, CONCURRENT_RUNS_PER_USER, HOSTED_AGENT_LIMITS
+from core.agent_versions import get_active_version
 from services.sandbox import compute_credits_for_run, get_provider
+from services.agent_redeploy import redeploy, RedeployError
+from services.agent_deps import build_agent_image, DepsError
 
 logger = logging.getLogger("wayforth")
 
@@ -319,6 +322,60 @@ async def _dispatch_run_internal(
     return run_id, credits_reserved
 
 
+def _versioned_dispatch_enabled() -> bool:
+    """Step 4 cutover flag (default OFF). When off, save + dispatch are the legacy
+    single-file .code paths, byte-identical to before."""
+    return os.environ.get("AGENT_VERSIONED_DISPATCH_ENABLED", "").strip().lower() in (
+        "1", "true", "yes", "on")
+
+
+async def _resolve_dispatch(pool, agent: dict) -> dict:
+    """Resolve what a run executes, ONCE per run — the in-flight isolation binding.
+
+    Returns {code, files, image_ref, version_id}. When the flag is off (or no active
+    version / no built image), this is the legacy .code path with version_id=None. When
+    the active version has a built image (python), the run boots that image + files. A
+    version without a built image (e.g. backfilled v1) still binds version_id for audit
+    but runs the legacy .code path — the NULL-image fallback.
+    """
+    async with pool.acquire() as conn:
+        code_row = await conn.fetchrow(
+            "SELECT code FROM hosted_agents WHERE id = $1::uuid", agent["id"])
+        code = (code_row or {}).get("code") or ""
+        if not _versioned_dispatch_enabled():
+            return {"code": code, "files": None, "image_ref": None, "version_id": None}
+        v = await get_active_version(conn, agent["id"])
+    if not v:
+        return {"code": code, "files": None, "image_ref": None, "version_id": None}
+    if not v.get("image_ref") or agent["runtime"] != "python3.12":
+        return {"code": code, "files": None, "image_ref": None, "version_id": v["id"]}
+    return {"code": code, "files": v["files"], "image_ref": v["image_ref"],
+            "version_id": v["id"]}
+
+
+def _mirror_config() -> tuple[str, str]:
+    """The private wheel mirror the build sandbox installs from (mirror-only egress).
+    Empty until the mirror stands up (Step 4b) — the flag stays off until then."""
+    url = os.environ.get("AGENT_MIRROR_URL", "").strip()
+    host = os.environ.get("AGENT_MIRROR_HOST", "").strip()
+    return url, host
+
+
+async def _prod_build_fn(*, agent: dict, version: dict, files: dict, requirements) -> str:
+    """Production build_fn for redeploy: the Step-1 egress-locked, wheels-only, hashed
+    build. Runs the sync build in an executor. Injected so tests use a fake instead."""
+    from services.agent_deps import build_requirements_lock  # local: avoid import cycle churn
+    mirror_url, mirror_host = _mirror_config()
+    reqs_text = build_requirements_lock(requirements) if requirements else ""
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(
+        None,
+        lambda: build_agent_image(
+            str(agent["id"]), version["version_no"], files, reqs_text,
+            mirror_url=mirror_url, mirror_host=mirror_host),
+    )
+
+
 async def _execute_run(
     pool,
     run_id: str,
@@ -354,7 +411,12 @@ async def _update(status: str, **fields) -> None:
             )
 
     try:
-        await _update("running", started_at=datetime.now(timezone.utc))
+        # Resolve the dispatch version ONCE and bind it to the run (in-flight isolation):
+        # a concurrent edit that repoints active_version_id afterward cannot change what
+        # this run executes. version_id stamped atomically with the 'running' transition.
+        dispatch = await _resolve_dispatch(pool, agent)
+        await _update("running", started_at=datetime.now(timezone.utc),
+                      version_id=dispatch["version_id"])
 
         # Decrypt user env vars at dispatch — decrypted value never stored or logged
         user_env: dict[str, str] = {}
@@ -368,12 +430,8 @@ async def _update(status: str, **fields) -> None:
             "WAYFORTH_PARAMS": json.dumps(params or {}),  # validated, resolved run params
         }
 
-        async with pool.acquire() as conn:
-            code_row = await conn.fetchrow(
-                "SELECT code FROM hosted_agents WHERE id = $1::uuid", agent["id"]
-            )
-        code = (code_row or {}).get("code") or ""
-        if not code.strip():
+        code = dispatch["code"]
+        if not (code.strip() or dispatch["files"]):
             if credits_reserved > 0:
                 await _release_reserve(pool, user_id, credits_reserved, run_id)
             await _update(
@@ -394,6 +452,8 @@ async def _update(status: str, **fields) -> None:
             runtime=agent["runtime"],
             env=run_env,
             timeout_seconds=min(timeout_s, _MAX_TIMEOUT),
+            files=dispatch["files"],
+            image_ref=dispatch["image_ref"],
         )
 
         # Deduct compute charge (1.5 credits/actual-min, ceil, min 1).
@@ -648,6 +708,21 @@ async def upload_code(
         code, json.dumps(params_schema) if params_schema is not None else None, agent_id,
     )
 
+    # Step 4: when versioned dispatch is on, also create→build→activate a version from
+    # this save (single-file entrypoint here; multi-file + requirements arrive in Step 5).
+    # The .code write above stays in sync as the NULL-image fallback. Flag off → legacy
+    # save only, behavior unchanged. A redeploy failure leaves the prior version serving.
+    if _versioned_dispatch_enabled():
+        from core.agent_versions import entrypoint_for_runtime
+        entrypoint = entrypoint_for_runtime(agent["runtime"])
+        try:
+            await redeploy(request.app.state.pool, dict(agent), {entrypoint: code}, "",
+                           build_fn=_prod_build_fn)
+        except RedeployError as e:
+            code_map = {"files": 422, "params": 422, "requirements": 422, "build": 502}
+            raise HTTPException(status_code=code_map.get(e.stage, 500), detail={
+                "error": f"redeploy_{e.stage}", "message": e.message, "errors": e.errors})
+
     ext = "ts" if agent["runtime"] == "node20" else "py"
     return {
         "id": agent_id,

apps/api/scripts/agent_redeploy_proof.py

@@ -0,0 +1,136 @@
+"""agent_redeploy_proof.py — code-editing v1, Step 4: real-DB proof of the orchestration.
+
+Runs the REAL services.agent_redeploy.redeploy + routers.cloud._resolve_dispatch against a
+live Postgres (point DATABASE_URL at a throwaway DB) with an injected fake build_fn — no
+E2B/mirror needed. Proves the guarantees the SQL layer is responsible for:
+
+  A. build failure → prior active version still serving (active pointer UNMOVED).
+  B. in-flight isolation → a run dispatched before an edit stays bound to its old version
+     (agent_runs.version_id), a run dispatched after gets the new one.
+  C. forward-only activate → a slow OLDER build cannot clobber a newer already-active version.
+
+Usage: DATABASE_URL=postgres://… AGENT_VERSIONED_DISPATCH_ENABLED=1 python scripts/agent_redeploy_proof.py
+"""
+import asyncio
+import json
+import os
+
+import asyncpg
+
+from services.agent_redeploy import redeploy, RedeployError
+from routers.cloud import _resolve_dispatch
+
+SCHEMA = """
+CREATE TABLE hosted_agents (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name TEXT, runtime TEXT DEFAULT 'python3.12',
+  code TEXT, params_schema JSONB, active_version_id UUID,
+  created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW());
+CREATE TABLE agent_versions (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(), agent_id UUID NOT NULL, version_no INT NOT NULL,
+  files JSONB NOT NULL, requirements JSONB, params_schema JSONB, image_ref TEXT,
+  status TEXT NOT NULL DEFAULT 'active', created_at TIMESTAMPTZ DEFAULT NOW(),
+  UNIQUE(agent_id, version_no));
+CREATE TABLE agent_runs (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(), agent_id UUID, version_id UUID, status TEXT,
+  created_at TIMESTAMPTZ DEFAULT NOW());
+"""
+
+CODE = "print('hello')"
+results = []
+
+
+def check(name, ok):
+    results.append((name, ok))
+    print(f"  {'PASS' if ok else 'FAIL'}  {name}")
+
+
+async def fake_build_ok(**k):
+    return f"img-v{k['version']['version_no']}"
+
+
+async def fake_build_fail(**k):
+    raise RuntimeError("install_failed: simulated bad build")
+
+
+async def dispatch_bind(pool, agent, run_status="running"):
+    """Mirror what _execute_run does: resolve the version once + stamp agent_runs.version_id."""
+    d = await _resolve_dispatch(pool, agent)
+    async with pool.acquire() as c:
+        rid = await c.fetchval(
+            "INSERT INTO agent_runs (agent_id, version_id, status) VALUES ($1::uuid,$2,$3) RETURNING id",
+            str(agent["id"]), d["version_id"], run_status)
+    return str(rid), d["version_id"]
+
+
+async def main():
+    pool = await asyncpg.create_pool(os.environ["DATABASE_URL"], min_size=1, max_size=4)
+    async with pool.acquire() as c:
+        await c.execute(SCHEMA)
+        agent_id = await c.fetchval(
+            "INSERT INTO hosted_agents (name, runtime, code) VALUES ('a','python3.12',$1) RETURNING id",
+            CODE)
+        # seed v1 active (mirrors the migration backfill)
+        v1 = await c.fetchval(
+            "INSERT INTO agent_versions (agent_id, version_no, files, status) "
+            "VALUES ($1, 1, $2, 'active') RETURNING id", agent_id, json.dumps({"agent.py": CODE}))
+        await c.execute("UPDATE hosted_agents SET active_version_id=$1 WHERE id=$2", v1, agent_id)
+    agent = {"id": str(agent_id), "runtime": "python3.12"}
+    v1 = str(v1)
+
+    # ── A. build failure → pointer UNMOVED ───────────────────────────────────────
+    print("A. build failure leaves prior active version serving")
+    try:
+        await redeploy(pool, agent, {"agent.py": CODE}, "", build_fn=fake_build_fail)
+        check("redeploy raises on build failure", False)
+    except RedeployError as e:
+        check("redeploy raises RedeployError(build)", e.stage == "build")
+    async with pool.acquire() as c:
+        active = str(await c.fetchval("SELECT active_version_id FROM hosted_agents WHERE id=$1", agent_id))
+        failed = await c.fetchval("SELECT count(*) FROM agent_versions WHERE agent_id=$1 AND status='failed'", agent_id)
+    check("active pointer still v1 (agent not broken)", active == v1)
+    check("failed build recorded as a 'failed' version", failed == 1)
+
+    # ── B. in-flight isolation via version_id binding ────────────────────────────
+    print("B. in-flight isolation (run before edit stays on old version)")
+    r1, r1_ver = await dispatch_bind(pool, agent)            # dispatched BEFORE the edit → binds v1
+    out = await redeploy(pool, agent, {"agent.py": CODE + "\n# edit"}, "", build_fn=fake_build_ok)
+    v2 = out["version_id"]
+    async with pool.acquire() as c:
+        active = str(await c.fetchval("SELECT active_version_id FROM hosted_agents WHERE id=$1", agent_id))
+    check("edit activated v2", active == v2 and out["activated"])
+    r2, r2_ver = await dispatch_bind(pool, agent)            # dispatched AFTER the edit → binds v2
+    check("run R1 (pre-edit) still bound to v1", r1_ver == v1)
+    check("run R2 (post-edit) bound to v2", r2_ver == v2)
+    check("the two runs resolved DIFFERENT versions", r1_ver != r2_ver)
+
+    # ── C. forward-only activate (older build can't clobber newer) ───────────────
+    print("C. forward-only activate")
+    async with pool.acquire() as c:
+        # current active is v2 (version_no=2). Forge an older built version (version_no=2's
+        # sibling at a lower number is impossible; instead make a higher-active then an older one)
+        vhi = await c.fetchval(
+            "INSERT INTO agent_versions (agent_id, version_no, files, image_ref, status) "
+            "VALUES ($1, 9, $2, 'img-v9', 'active') RETURNING id", agent_id, json.dumps({"agent.py": CODE}))
+        await c.execute("UPDATE hosted_agents SET active_version_id=$1 WHERE id=$2", vhi, agent_id)
+        vlo = await c.fetchval(
+            "INSERT INTO agent_versions (agent_id, version_no, files, image_ref, status) "
+            "VALUES ($1, 8, $2, 'img-v8', 'building') RETURNING id", agent_id, json.dumps({"agent.py": CODE}))
+        # the forward-only activate (same SQL the orchestrator uses) for the OLDER version
+        moved = await c.fetchval(
+            """UPDATE hosted_agents h SET active_version_id=$2::uuid
+               WHERE h.id=$1::uuid
+                 AND (h.active_version_id IS NULL OR
+                      (SELECT version_no FROM agent_versions WHERE id=h.active_version_id) < $3)
+               RETURNING h.active_version_id""",
+            str(agent_id), str(vlo), 8)
+        active = str(await c.fetchval("SELECT active_version_id FROM hosted_agents WHERE id=$1", agent_id))
+    check("older build (v8) did NOT clobber newer active (v9)", moved is None and active == str(vhi))
+
+    await pool.close()
+    ok = all(v for _, v in results)
+    print(f"\n{'ALL PROPERTIES PROVEN ✓' if ok else 'FAILURES ✗'}  ({sum(v for _,v in results)}/{len(results)})")
+    raise SystemExit(0 if ok else 1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())