feat(code-editing): Step 5 — rollback + multi-file deploy + revocation flagging

apps/api/core/agent_versions.py

@@ -92,3 +92,27 @@ async def activate_version(db, agent_id, version_id) -> None:
            RETURNING h.id""", str(agent_id), str(version_id))
     if not updated:
         raise ValueError(f"version {version_id} does not belong to agent {agent_id}")
+
+
+async def rollback_to(conn, agent_id, version_id) -> dict:
+    """Roll the active version back to a prior one — instant and safe: its image is
+    already built, so this is a pure pointer repoint (no rebuild). Deliberately NOT
+    forward-only (rollback goes backward by design). Rejects rolling back to a
+    failed/building version (nothing usable to serve). Run inside a transaction."""
+    target = await conn.fetchrow(
+        "SELECT version_no, status FROM agent_versions WHERE id = $1::uuid AND agent_id = $2::uuid",
+        str(version_id), str(agent_id))
+    if not target:
+        raise ValueError(f"version {version_id} does not belong to agent {agent_id}")
+    if target["status"] in ("failed", "building"):
+        raise ValueError(f"cannot roll back to a '{target['status']}' version")
+    await conn.execute(
+        "UPDATE hosted_agents SET active_version_id = $2::uuid, updated_at = NOW() "
+        "WHERE id = $1::uuid", str(agent_id), str(version_id))
+    await conn.execute(
+        """UPDATE agent_versions
+           SET status = CASE WHEN id = $2::uuid THEN 'active'
+                             WHEN status = 'active' THEN 'superseded' ELSE status END
+           WHERE agent_id = $1::uuid AND (id = $2::uuid OR status = 'active')""",
+        str(agent_id), str(version_id))
+    return {"version_id": str(version_id), "version_no": target["version_no"]}

apps/api/core/package_revocation.py

@@ -0,0 +1,56 @@
+"""core/package_revocation.py — code-editing v1, Step 5: package revocation flagging.
+
+When an allowlisted package is later found malicious/vulnerable, revoke_package records
+it and FLAGS every agent_version whose baked requirements include it (so affected agents
+can be rebuilt). is_revoked / revoked_pins let the redeploy orchestrator block NEW builds
+from using a revoked package.
+
+agent_versions.requirements is a JSON array of [name, version, [hashes]] (the validated
+pins), so an element's ->>0 is the package name and ->>1 its version.
+"""
+from __future__ import annotations
+
+
+async def revoke_package(conn, name: str, *, version: str | None = None,
+                         reason: str | None = None) -> int:
+    """Record the revocation and flag affected versions. Returns the count flagged.
+    version=None revokes ALL versions of the package."""
+    await conn.execute(
+        """INSERT INTO revoked_packages (name, version, reason) VALUES ($1, $2, $3)
+           ON CONFLICT (name, version) DO UPDATE SET reason = EXCLUDED.reason, revoked_at = NOW()""",
+        name, version, reason)
+    rows = await conn.fetch(
+        """UPDATE agent_versions SET dep_flagged = TRUE
+           WHERE requirements IS NOT NULL
+             AND EXISTS (SELECT 1 FROM jsonb_array_elements(requirements) e
+                         WHERE e->>0 = $1 AND ($2::text IS NULL OR e->>1 = $2))
+           RETURNING id""", name, version)
+    return len(rows)
+
+
+async def is_revoked(conn, name: str, version: str | None = None) -> bool:
+    """True if (name, version) is revoked — either a blanket revocation (version NULL)
+    or one matching this exact version."""
+    return bool(await conn.fetchval(
+        """SELECT 1 FROM revoked_packages
+           WHERE name = $1 AND (version IS NULL OR version = $2) LIMIT 1""",
+        name, version))
+
+
+async def revoked_pins(conn, pins) -> list:
+    """Subset of pins (name, version, …) that are revoked — for redeploy rejection."""
+    bad = []
+    for p in pins:
+        if await is_revoked(conn, p[0], p[1]):
+            bad.append((p[0], p[1]))
+    return bad
+
+
+async def flagged_versions(conn) -> list:
+    """All currently dep_flagged versions (agent_id, version_id, version_no) — for ops
+    to find + rebuild affected agents."""
+    rows = await conn.fetch(
+        """SELECT agent_id, id AS version_id, version_no FROM agent_versions
+           WHERE dep_flagged ORDER BY agent_id, version_no""")
+    return [{"agent_id": str(r["agent_id"]), "version_id": str(r["version_id"]),
+             "version_no": r["version_no"]} for r in rows]

apps/api/main.py

@@ -1200,6 +1200,22 @@ async def lifespan(app: FastAPI):
             await _mconn.execute("""
                 CREATE INDEX IF NOT EXISTS idx_agent_runs_version ON agent_runs(version_id)
             """)
+            # Migration 071: package revocation flagging (code-editing Step 5).
+            # Mirrored in infra/migrations/071_package_revocation.sql. Additive/reversible.
+            await _mconn.execute("""
+                CREATE TABLE IF NOT EXISTS revoked_packages (
+                    name TEXT NOT NULL, version TEXT, reason TEXT,
+                    revoked_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE (name, version)
+                )
+            """)
+            await _mconn.execute("""
+                ALTER TABLE agent_versions
+                    ADD COLUMN IF NOT EXISTS dep_flagged BOOLEAN NOT NULL DEFAULT FALSE
+            """)
+            await _mconn.execute("""
+                CREATE INDEX IF NOT EXISTS idx_agent_versions_dep_flagged
+                    ON agent_versions(agent_id) WHERE dep_flagged
+            """)
             await _mconn.execute("""
                 CREATE INDEX IF NOT EXISTS idx_hosted_agents_next_run
                     ON hosted_agents(next_run_at)

apps/api/pytest.ini

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

apps/api/routers/admin/__init__.py

@@ -14,6 +14,7 @@
 from .services import router as services_router
 from .dashboard import router as dashboard_router
 from .usdc import router as usdc_router
+from .packages import router as packages_router
 
 logger = logging.getLogger("wayforth")
 
@@ -275,3 +276,4 @@ async def admin_page():
 router.include_router(services_router)
 router.include_router(dashboard_router)
 router.include_router(usdc_router)
+router.include_router(packages_router)

apps/api/routers/admin/packages.py

@@ -0,0 +1,38 @@
+"""routers/admin/packages.py — /admin/packages/* (package revocation flagging, Step 5)."""
+import logging
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse
+
+from core.admin_auth import admin_authed
+from core.db import get_db
+from core.package_revocation import flagged_versions, revoke_package
+
+router = APIRouter()
+logger = logging.getLogger("wayforth")
+
+
+@router.post("/admin/packages/revoke", tags=["Admin"])
+async def revoke(request: Request, db=Depends(get_db)):
+    """Revoke a (previously allowlisted) package and flag every version that baked it in.
+    Body: {name, version?, reason?}. version omitted ⇒ all versions revoked."""
+    if not await admin_authed(request, db):
+        return JSONResponse({"error": "unauthorized"}, status_code=401)
+    body = await request.json()
+    name = (body.get("name") or "").strip()
+    if not name:
+        return JSONResponse({"error": "name required"}, status_code=422)
+    version, reason = body.get("version"), body.get("reason")
+    async with db.transaction():
+        flagged = await revoke_package(db, name, version=version, reason=reason)
+    logger.warning("package revoked: %s%s (%d versions flagged)", name,
+                   f"=={version}" if version else " (all versions)", flagged)
+    return {"revoked": name, "version": version, "versions_flagged": flagged}
+
+
+@router.get("/admin/packages/flagged", tags=["Admin"])
+async def flagged(request: Request, db=Depends(get_db)):
+    """Versions currently flagged as using a revoked package — for rebuild/triage."""
+    if not await admin_authed(request, db):
+        return JSONResponse({"error": "unauthorized"}, status_code=401)
+    return {"flagged_versions": await flagged_versions(db)}

apps/api/routers/cloud.py

@@ -46,7 +46,7 @@
 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 core.agent_versions import get_active_version, list_versions, rollback_to
 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
@@ -719,9 +719,7 @@ async def upload_code(
             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})
+            raise _redeploy_http(e)
 
     ext = "ts" if agent["runtime"] == "node20" else "py"
     return {
@@ -735,6 +733,93 @@ async def upload_code(
     }
 
 
+def _redeploy_http(e: RedeployError) -> HTTPException:
+    """Map a RedeployError stage to an HTTP status (validation → 422, build → 502)."""
+    code_map = {"files": 422, "params": 422, "requirements": 422, "build": 502}
+    return HTTPException(status_code=code_map.get(e.stage, 500), detail={
+        "error": f"redeploy_{e.stage}", "message": e.message, "errors": e.errors})
+
+
+@router.post("/agents/{agent_id}/deploy", status_code=200)
+@limiter.limit("30/minute")
+async def deploy_version(request: Request, agent_id: str, db=Depends(get_db)) -> dict:
+    """Multi-file versioned deploy: {files: {path: content}, requirements: "name==ver\\n…"}.
+    Runs the full save→redeploy (create→validate→build→activate). Flag-gated."""
+    user_id, _, tier = await _resolve_caller(request, db)
+    require_tier(tier, "cloud_agents")
+    if not _versioned_dispatch_enabled():
+        raise HTTPException(status_code=409, detail={"error": "versioning_disabled",
+            "message": "Versioned multi-file deploy is not enabled."})
+    agent = await _get_agent_or_404(db, user_id, agent_id)
+    if agent["status"] == "running":
+        raise HTTPException(status_code=409, detail={"error": "agent_running",
+            "message": "Cannot deploy while a run is in progress."})
+    body = await request.json()
+    files = body.get("files") or {}
+    requirements = body.get("requirements") or ""
+    if not isinstance(files, dict) or not files:
+        raise HTTPException(status_code=422, detail={"error": "no_files",
+            "message": "A non-empty 'files' map is required."})
+    total = sum(len(c.encode("utf-8")) for c in files.values() if isinstance(c, str))
+    if total > _MAX_CODE_BYTES:
+        raise HTTPException(status_code=413, detail={"error": "code_too_large",
+            "max_bytes": _MAX_CODE_BYTES})
+    try:
+        out = await redeploy(request.app.state.pool, dict(agent), files, requirements,
+                             build_fn=_prod_build_fn)
+    except RedeployError as e:
+        raise _redeploy_http(e)
+    return {"id": agent_id, "version_no": out["version_no"], "version_id": out["version_id"],
+            "image_ref": out["image_ref"], "activated": out["activated"], "status": out["status"]}
+
+
+@router.get("/agents/{agent_id}/versions")
+@limiter.limit("60/minute")
+async def list_agent_versions(request: Request, agent_id: str, db=Depends(get_db)) -> dict:
+    """Version history (newest first) + the active pointer — what rollback can target."""
+    user_id, _, tier = await _resolve_caller(request, db)
+    require_tier(tier, "cloud_agents")
+    agent = await _get_agent_or_404(db, user_id, agent_id)
+    versions = await list_versions(db, agent_id)
+    active = agent.get("active_version_id")
+    return {"id": agent_id, "active_version_id": str(active) if active else None,
+            "versions": versions}
+
+
+@router.post("/agents/{agent_id}/rollback", status_code=200)
+@limiter.limit("30/minute")
+async def rollback_version(request: Request, agent_id: str, db=Depends(get_db)) -> dict:
+    """Roll the active version back to a prior one (by version_id or version_no). Instant
+    + safe — the target's image is already built, so this is a pure pointer repoint."""
+    user_id, _, tier = await _resolve_caller(request, db)
+    require_tier(tier, "cloud_agents")
+    if not _versioned_dispatch_enabled():
+        raise HTTPException(status_code=409, detail={"error": "versioning_disabled",
+            "message": "Rollback requires versioned dispatch."})
+    agent = await _get_agent_or_404(db, user_id, agent_id)
+    if agent["status"] == "running":
+        raise HTTPException(status_code=409, detail={"error": "agent_running",
+            "message": "Cannot roll back while a run is in progress."})
+    body = await request.json()
+    version_id, version_no = body.get("version_id"), body.get("version_no")
+    pool = request.app.state.pool
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            if version_id is None and version_no is not None:
+                version_id = await conn.fetchval(
+                    "SELECT id FROM agent_versions WHERE agent_id = $1::uuid AND version_no = $2",
+                    agent_id, int(version_no))
+            if not version_id:
+                raise HTTPException(status_code=404, detail={"error": "version_not_found",
+                    "message": "Target version not found for this agent."})
+            try:
+                out = await rollback_to(conn, agent_id, str(version_id))
+            except ValueError as e:
+                raise HTTPException(status_code=422, detail={"error": "rollback_rejected",
+                    "message": str(e)})
+    return {"id": agent_id, "rolled_back_to": out["version_no"], "version_id": out["version_id"]}
+
+
 @router.get("/agents")
 @limiter.limit("60/minute")
 async def list_agents(request: Request, db=Depends(get_db)) -> dict:

apps/api/scripts/agent_redeploy_proof.py

@@ -19,6 +19,8 @@
 
 from services.agent_redeploy import redeploy, RedeployError
 from routers.cloud import _resolve_dispatch
+from core.agent_versions import rollback_to
+from core.package_revocation import revoke_package, is_revoked, flagged_versions
 
 SCHEMA = """
 CREATE TABLE hosted_agents (
@@ -28,11 +30,14 @@
 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));
+  status TEXT NOT NULL DEFAULT 'active', dep_flagged BOOLEAN NOT NULL DEFAULT FALSE,
+  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());
+CREATE TABLE revoked_packages (
+  name TEXT NOT NULL, version TEXT, reason TEXT,
+  revoked_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE(name, version));
 """
 
 CODE = "print('hello')"
@@ -126,6 +131,49 @@ async def main():
         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))
 
+    # ── D. rollback — instant pointer repoint to a prior built version ───────────
+    print("D. rollback to a prior version (no rebuild)")
+    async with pool.acquire() as c:
+        async with c.transaction():
+            out = await rollback_to(c, str(agent_id), v1)        # active is v9 → roll back to v1
+        active = str(await c.fetchval("SELECT active_version_id FROM hosted_agents WHERE id=$1", agent_id))
+        v1_status = await c.fetchval("SELECT status FROM agent_versions WHERE id=$1::uuid", v1)
+    check("rollback repointed active to v1", active == v1 and out["version_no"] == 1)
+    check("rolled-back version is now active", v1_status == "active")
+    # rollback to a failed version is rejected
+    async with pool.acquire() as c:
+        bad = await c.fetchval(
+            "INSERT INTO agent_versions (agent_id, version_no, files, status) "
+            "VALUES ($1, 99, $2, 'failed') RETURNING id", agent_id, json.dumps({"agent.py": CODE}))
+        try:
+            async with c.transaction():
+                await rollback_to(c, str(agent_id), str(bad))
+            check("rollback to a 'failed' version rejected", False)
+        except ValueError:
+            check("rollback to a 'failed' version rejected", True)
+
+    # ── E. package revocation flagging ──────────────────────────────────────────
+    print("E. package revocation flags affected versions")
+    async with pool.acquire() as c:
+        vr = await c.fetchval(
+            "INSERT INTO agent_versions (agent_id, version_no, files, requirements, status) "
+            "VALUES ($1, 50, $2, $3, 'active') RETURNING id", agent_id,
+            json.dumps({"agent.py": CODE}), json.dumps([["badpkg", "1.0", ["sha256:x"]]]))
+        v_safe = await c.fetchval(
+            "INSERT INTO agent_versions (agent_id, version_no, files, requirements, status) "
+            "VALUES ($1, 51, $2, $3, 'active') RETURNING id", agent_id,
+            json.dumps({"agent.py": CODE}), json.dumps([["httpx", "0.28.1", ["sha256:y"]]]))
+        async with c.transaction():
+            n = await revoke_package(c, "badpkg", version="1.0", reason="canary")
+        flagged_bad = await c.fetchval("SELECT dep_flagged FROM agent_versions WHERE id=$1::uuid", vr)
+        flagged_safe = await c.fetchval("SELECT dep_flagged FROM agent_versions WHERE id=$1::uuid", v_safe)
+        revoked = await is_revoked(c, "badpkg", "1.0")
+        flist = await flagged_versions(c)
+    check("revoke flagged exactly the version using badpkg", n == 1 and flagged_bad is True)
+    check("unrelated version (httpx) NOT flagged", flagged_safe is False)
+    check("is_revoked(badpkg==1.0) true", revoked is True)
+    check("flagged_versions surfaces the affected version", any(f["version_id"] == str(vr) for f in flist))
+
     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)})")

apps/api/services/agent_redeploy.py

@@ -86,6 +86,15 @@ async def redeploy(pool, agent: dict, files: dict, requirements_text: str, *, bu
     pins, errors = validate_requirements(requirements_text or "")
     if errors:
         raise RedeployError("requirements", "requirements rejected", errors=errors)
+    # a package can be allowlisted yet later REVOKED — block new builds that use one
+    if pins:
+        from core.package_revocation import revoked_pins
+        async with pool.acquire() as conn:
+            bad = await revoked_pins(conn, pins)
+        if bad:
+            raise RedeployError("requirements", "revoked package(s)", errors=[
+                {"field": n, "code": "revoked", "message": f"'{n}=={v}' is revoked"}
+                for n, v in bad])
 
     # ── create the building version (immutable attempt record) ───────────────────
     version = await _create_building_version(pool, agent_id, files, pins, params_schema)