diff --git a/apps/api/main.py b/apps/api/main.py index 33a5d88..6c05054 100644 --- a/apps/api/main.py +++ b/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) diff --git a/apps/api/pytest.ini b/apps/api/pytest.ini index aff24d7..2595d35 100644 --- a/apps/api/pytest.ini +++ b/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) diff --git a/apps/api/routers/cloud.py b/apps/api/routers/cloud.py index bbc594f..1c69eee 100644 --- a/apps/api/routers/cloud.py +++ b/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, diff --git a/apps/api/scripts/agent_redeploy_proof.py b/apps/api/scripts/agent_redeploy_proof.py new file mode 100644 index 0000000..5c5cbf2 --- /dev/null +++ b/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()) diff --git a/apps/api/services/agent_redeploy.py b/apps/api/services/agent_redeploy.py new file mode 100644 index 0000000..1a57997 --- /dev/null +++ b/apps/api/services/agent_redeploy.py @@ -0,0 +1,131 @@ +"""services/agent_redeploy.py — code-editing v1, Step 4: save→redeploy orchestration. + +create version → validate (PARAMS extraction + requirements vs allowlist) → build image +(injectable build_fn — the Step-1 egress-locked pipeline in prod, a fake in tests) → +forward-only atomic activate. + +Invariants (the ones reviewed hardest): + • FAIL-CLOSED: any failure before activation leaves the prior active version serving — + the active pointer never moves unless a fully-built, validated image exists. + • FORWARD-ONLY ACTIVATE: activation only advances to a higher version_no, so a slow + older build can never clobber a newer already-active version (last-submitted wins) + WITHOUT holding a lock/transaction across the multi-second build. + • No lock is held on the dispatch path — in-flight isolation comes from dispatch + resolving the active version once and stamping agent_runs.version_id (Step 4 wiring). + +Call sites are flag-gated (AGENT_VERSIONED_DISPATCH_ENABLED) — inert until flip + the +real mirror stands up (Step 4b). +""" +from __future__ import annotations + +import logging + +from asyncpg.exceptions import UniqueViolationError + +from core.agent_versions import create_version, entrypoint_for_runtime +from core.params_schema import ParamsSchemaError, compile_params +from services.agent_deps import validate_requirements + +logger = logging.getLogger("wayforth") + + +class RedeployError(Exception): + """A redeploy failed at a named stage. The active version is unchanged.""" + def __init__(self, stage: str, message: str, errors: list | None = None): + super().__init__(message) + self.stage = stage # 'files' | 'params' | 'requirements' | 'build' + self.message = message + self.errors = errors or [] + + +def _extract_params_schema(runtime: str, entrypoint_code: str): + """Static PARAMS extraction (python only; never executes code). Mirrors the upload + path so a save and a redeploy validate identically.""" + if (runtime or "").startswith("python"): + return compile_params(entrypoint_code, "python") + return None + + +async def _create_building_version(pool, agent_id, files, pins, params_schema, *, tries=4): + """Insert a building version; retry on the version_no race (two concurrent saves + both compute MAX+1 → one hits UNIQUE(agent_id, version_no) → recompute + retry).""" + for attempt in range(tries): + try: + async with pool.acquire() as conn: + return await create_version( + conn, agent_id, files, requirements=pins, + params_schema=params_schema, status="building") + except UniqueViolationError: + if attempt == tries - 1: + raise + logger.info("redeploy: version_no race for agent=%s, retrying", agent_id) + + +async def _mark_version(pool, version_id, status, error=None): + async with pool.acquire() as conn: + await conn.execute( + "UPDATE agent_versions SET status = $2 WHERE id = $1::uuid", str(version_id), status) + if error: + logger.warning("redeploy: version %s -> %s: %s", version_id, status, error) + + +async def redeploy(pool, agent: dict, files: dict, requirements_text: str, *, build_fn) -> dict: + """Run the full save→redeploy. Returns the version + activation outcome, or raises + RedeployError (prior active version still serving).""" + agent_id = str(agent["id"]) + runtime = agent["runtime"] + entrypoint = entrypoint_for_runtime(runtime) + + # ── validate (no DB writes; fail-closed BEFORE spending a build) ────────────── + if entrypoint not in files: + raise RedeployError("files", f"missing entrypoint '{entrypoint}'") + try: + params_schema = _extract_params_schema(runtime, files[entrypoint]) + except ParamsSchemaError as e: + raise RedeployError("params", str(e)) + pins, errors = validate_requirements(requirements_text or "") + if errors: + raise RedeployError("requirements", "requirements rejected", errors=errors) + + # ── create the building version (immutable attempt record) ─────────────────── + version = await _create_building_version(pool, agent_id, files, pins, params_schema) + vid, vno = version["id"], version["version_no"] + + # ── build the image (injectable; NO db txn/lock held across the slow build) ── + try: + image_ref = await build_fn(agent=agent, version=version, files=files, requirements=pins) + except Exception as e: # build/validation failure inside the pipeline + await _mark_version(pool, vid, "failed", error=str(e)) + raise RedeployError("build", str(e)) + + # ── forward-only atomic activate ───────────────────────────────────────────── + async with pool.acquire() as conn: + async with conn.transaction(): + await conn.execute( + "UPDATE agent_versions SET image_ref = $2 WHERE id = $1::uuid", vid, image_ref) + moved = await conn.fetchval( + """UPDATE hosted_agents h + SET active_version_id = $2::uuid, updated_at = NOW() + WHERE h.id = $1::uuid + AND EXISTS (SELECT 1 FROM agent_versions av + WHERE av.id = $2::uuid AND av.agent_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""", + agent_id, vid, vno) + if moved: + # new version active; demote whatever was active before + await conn.execute( + """UPDATE agent_versions + SET status = CASE WHEN id = $2::uuid THEN 'active' ELSE 'superseded' END + WHERE agent_id = $1::uuid AND (id = $2::uuid OR status = 'active')""", + agent_id, vid) + else: + # lost the forward-only race to a newer version — built but never active + await conn.execute( + "UPDATE agent_versions SET status = 'superseded' " + "WHERE id = $1::uuid AND status = 'building'", vid) + + return {"version_id": vid, "version_no": vno, "image_ref": image_ref, + "activated": bool(moved), "status": "active" if moved else "superseded"} diff --git a/apps/api/services/sandbox.py b/apps/api/services/sandbox.py index 20ea7ff..443ce29 100644 --- a/apps/api/services/sandbox.py +++ b/apps/api/services/sandbox.py @@ -78,6 +78,9 @@ async def run( runtime: str, # 'python3.12' | 'node20' env: dict[str, str], timeout_seconds: int, + *, + files: dict[str, str] | None = None, # Step 4: multi-file version (path: content) + image_ref: str | None = None, # Step 4: per-version built image to boot ) -> SandboxResult: ... @@ -99,10 +102,13 @@ async def run( runtime: str, env: dict[str, str], timeout_seconds: int, + *, + files: dict[str, str] | None = None, + image_ref: str | None = None, ) -> SandboxResult: loop = asyncio.get_event_loop() return await loop.run_in_executor( - None, self._run_sync, code, runtime, env, timeout_seconds + None, self._run_sync, code, runtime, env, timeout_seconds, files, image_ref ) def _run_sync( @@ -111,9 +117,17 @@ def _run_sync( runtime: str, env: dict[str, str], timeout_seconds: int, + files: dict[str, str] | None = None, + image_ref: str | None = None, ) -> SandboxResult: from e2b import Sandbox, SandboxNetworkOpts + # Step 4 versioned dispatch: a per-version image (image_ref) carries baked deps, + # so it boots that image with gateway-ONLY egress and writes the version's files — + # no run-time pip. Set only when AGENT_VERSIONED_DISPATCH_ENABLED is on AND the + # version has a built image (callers fall back to the code path otherwise). + versioned = bool(image_ref) + # 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 = ( @@ -126,7 +140,11 @@ def _run_sync( "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: + if versioned: + create_kwargs["network"] = SandboxNetworkOpts( + deny_out=["0.0.0.0/0"], allow_out=[_GATEWAY_HOST]) + create_kwargs["template"] = image_ref + elif python_locked: create_kwargs["network"] = SandboxNetworkOpts( deny_out=["0.0.0.0/0"], allow_out=[_GATEWAY_HOST]) create_kwargs["template"] = _agent_base_image() @@ -138,9 +156,11 @@ def _run_sync( sandbox_id = sbx.sandbox_id try: if runtime == "python3.12": - sbx.files.write("/home/user/agent.py", code) - if python_locked: - # deps are baked into the base image — no run-time pip + # multi-file version writes all files; single-file path writes agent.py + for path, content in (files or {"agent.py": code}).items(): + sbx.files.write(f"/home/user/{path}", content) + if versioned or python_locked: + # deps are baked into the image — no run-time pip cmd = "python3 /home/user/agent.py" else: cmd = ( diff --git a/apps/api/tests/test_agent_redeploy.py b/apps/api/tests/test_agent_redeploy.py new file mode 100644 index 0000000..0a503bd --- /dev/null +++ b/apps/api/tests/test_agent_redeploy.py @@ -0,0 +1,200 @@ +"""test_agent_redeploy.py — code-editing v1, Step 4 (save→redeploy orchestration). + +Control-flow unit tests with a fake pool: validation is fail-closed BEFORE a build is +spent, and a build failure never reaches the activate step (pointer untouched). The +SQL-level guarantees (pointer-unmoved, version_id binding, forward-only activate) are +proven against real Postgres in scripts/agent_redeploy_proof.py. +""" +from __future__ import annotations + +import json + +import pytest + +from services import agent_redeploy as R +from services.agent_redeploy import RedeployError + + +class _Txn: + async def __aenter__(self): return self + async def __aexit__(self, *a): return False + + +class FakeConn: + def __init__(self, store): + self.store = store + + async def fetchval(self, q, *a): + self.store["queries"].append(q) + if "COALESCE(MAX(version_no)" in q: + return self.store["next_no"] + if "UPDATE hosted_agents" in q and "active_version_id" in q: + self.store["activated"] = True # forward-only activate fired + return "agent-id" + return None + + async def fetchrow(self, q, *a): + self.store["queries"].append(q) + if "INSERT INTO agent_versions" in q: + self.store["created"] = True + return {"id": f"vid-{a[1]}", "agent_id": a[0], "version_no": a[1], + "files": a[2], "requirements": a[3], "params_schema": a[4], + "image_ref": a[5], "status": a[6]} + return None + + async def execute(self, q, *a): + self.store["queries"].append(q) + return "OK" + + def transaction(self): return _Txn() + + +class _Acq: + def __init__(self, store): self.store = store + async def __aenter__(self): return FakeConn(self.store) + async def __aexit__(self, *a): return False + + +class FakePool: + def __init__(self, store): self.store = store + def acquire(self): return _Acq(self.store) + + +def _store(): + return {"queries": [], "next_no": 2, "created": False, "activated": False} + + +AGENT = {"id": "agent-1", "runtime": "python3.12"} +GOOD_CODE = "print('hi')" # no PARAMS declared → schema None, valid + + +def _activate_ran(store): + return any("UPDATE hosted_agents" in q and "active_version_id" in q for q in store["queries"]) + + +def _insert_ran(store): + return any("INSERT INTO agent_versions" in q for q in store["queries"]) + + +# ── validation is fail-closed BEFORE any build/version ────────────────────────── + +@pytest.mark.asyncio +async def test_rejects_missing_entrypoint(): + store = _store() + called = {"build": False} + async def build_fn(**k): called["build"] = True; return "img" + with pytest.raises(RedeployError) as ei: + await R.redeploy(FakePool(store), AGENT, {"helper.py": "x"}, "", build_fn=build_fn) + assert ei.value.stage == "files" + assert not called["build"] and not _insert_ran(store) # nothing spent + + +@pytest.mark.asyncio +async def test_rejects_non_literal_params(): + store = _store() + called = {"build": False} + async def build_fn(**k): called["build"] = True; return "img" + bad = "PARAMS = build_params()\nprint(1)" # non-literal → ParamsSchemaError + with pytest.raises(RedeployError) as ei: + await R.redeploy(FakePool(store), AGENT, {"agent.py": bad}, "", build_fn=build_fn) + assert ei.value.stage == "params" + assert not called["build"] and not _insert_ran(store) + + +@pytest.mark.asyncio +async def test_rejects_non_allowlisted_requirement(): + store = _store() + called = {"build": False} + async def build_fn(**k): called["build"] = True; return "img" + with pytest.raises(RedeployError) as ei: + await R.redeploy(FakePool(store), AGENT, {"agent.py": GOOD_CODE}, + "evil-canary==1.0", build_fn=build_fn) + assert ei.value.stage == "requirements" + assert ei.value.errors and not called["build"] and not _insert_ran(store) + + +# ── build failure: version recorded failed, ACTIVATE NEVER RUNS (pointer unmoved) ─ + +@pytest.mark.asyncio +async def test_build_failure_marks_failed_and_never_activates(): + store = _store() + async def build_fn(**k): raise RuntimeError("install_failed: boom") + with pytest.raises(RedeployError) as ei: + await R.redeploy(FakePool(store), AGENT, {"agent.py": GOOD_CODE}, "", build_fn=build_fn) + assert ei.value.stage == "build" + assert _insert_ran(store) # version was created + assert any("SET status = $2" in q for q in store["queries"]) # marked (failed) + assert not _activate_ran(store) # the pointer is NEVER touched + + +# ── success: build then forward-only activate fires ───────────────────────────── + +@pytest.mark.asyncio +async def test_success_builds_then_activates(): + store = _store() + seen = {} + async def build_fn(**k): seen.update(k); return "img-123" + out = await R.redeploy(FakePool(store), AGENT, {"agent.py": GOOD_CODE}, "", build_fn=build_fn) + assert seen["files"] == {"agent.py": GOOD_CODE} # build_fn got the files + assert _activate_ran(store) and out["activated"] and out["status"] == "active" + assert out["image_ref"] == "img-123" + + +# ── dispatch resolution: flag OFF is byte-identical to the legacy .code path ───── + +class _DispatchConn: + """Returns the code row for the legacy read; None for the active-version lookup.""" + def __init__(self, code="hi", version=None): + self.code, self.version = code, version + async def fetchrow(self, q, *a): + if "SELECT code FROM hosted_agents" in q: + return {"code": self.code} + return self.version # get_active_version's row + async def fetchval(self, q, *a): + return self.version["id"] if self.version else None + + +class _DispatchPool: + def __init__(self, conn): self._c = conn + def acquire(self): + c = self._c + class _A: + async def __aenter__(self): return c + async def __aexit__(self, *a): return False + return _A() + + +@pytest.mark.asyncio +async def test_resolve_dispatch_flag_off_is_legacy(monkeypatch): + monkeypatch.delenv("AGENT_VERSIONED_DISPATCH_ENABLED", raising=False) + from routers.cloud import _resolve_dispatch + d = await _resolve_dispatch(_DispatchPool(_DispatchConn(code="legacy code")), AGENT) + assert d == {"code": "legacy code", "files": None, "image_ref": None, "version_id": None} + + +@pytest.mark.asyncio +async def test_resolve_dispatch_flag_on_no_active_version_falls_back(monkeypatch): + monkeypatch.setenv("AGENT_VERSIONED_DISPATCH_ENABLED", "1") + from routers.cloud import _resolve_dispatch + d = await _resolve_dispatch(_DispatchPool(_DispatchConn(code="x", version=None)), AGENT) + assert d["files"] is None and d["image_ref"] is None and d["version_id"] is None + + +@pytest.mark.asyncio +async def test_resolve_dispatch_null_image_binds_but_uses_code(monkeypatch): + monkeypatch.setenv("AGENT_VERSIONED_DISPATCH_ENABLED", "1") + from routers.cloud import _resolve_dispatch + ver = {"id": "v-backfilled", "files": {"agent.py": "x"}, "image_ref": None} + d = await _resolve_dispatch(_DispatchPool(_DispatchConn(code="x", version=ver)), AGENT) + # version exists but no built image → legacy code path, version_id still bound for audit + assert d["version_id"] == "v-backfilled" and d["image_ref"] is None and d["files"] is None + + +@pytest.mark.asyncio +async def test_resolve_dispatch_built_image_uses_versioned_path(monkeypatch): + monkeypatch.setenv("AGENT_VERSIONED_DISPATCH_ENABLED", "1") + from routers.cloud import _resolve_dispatch + ver = {"id": "v2", "files": {"agent.py": "y", "helper.py": "z"}, "image_ref": "img-v2"} + d = await _resolve_dispatch(_DispatchPool(_DispatchConn(code="x", version=ver)), AGENT) + assert d["image_ref"] == "img-v2" and d["files"] == {"agent.py": "y", "helper.py": "z"} + assert d["version_id"] == "v2" diff --git a/infra/migrations/070_agent_run_version.sql b/infra/migrations/070_agent_run_version.sql new file mode 100644 index 0000000..7637b85 --- /dev/null +++ b/infra/migrations/070_agent_run_version.sql @@ -0,0 +1,11 @@ +-- 070_agent_run_version.sql — code-editing v1, Step 4: bind each run to its version. +-- +-- Records which agent_version a run executed. This is the in-flight isolation anchor: +-- dispatch resolves the active version ONCE at start and stamps version_id here, so a +-- later edit (which repoints active_version_id) cannot change a running run's binding. +-- Additive + nullable; dispatch is unchanged until AGENT_VERSIONED_DISPATCH_ENABLED flips. + +ALTER TABLE agent_runs + ADD COLUMN IF NOT EXISTS version_id UUID REFERENCES agent_versions(id); + +CREATE INDEX IF NOT EXISTS idx_agent_runs_version ON agent_runs(version_id);