"""routes_web_agent.py — the door that lets a person TEST the web agent (wave 31, R10 / D-51). The owner's words are the whole reason this file exists: *"we already laid the foundation of this but never test anything."* The capability is otherwise reachable only from inside an automation run, which means the first person to discover it is broken is a customer at 3am. GET /api/v1/web-agent/capability any session — can this deployment run a web step, and if not, the SENTENCE saying why POST /api/v1/web-agent/test ADMIN — run one real `web_read` and show what came back, or the sentence ⛔ NEITHER ROUTE IS THE SEAM. `automation_engine` calls `web_agent.run_step` directly (contract C5); these are an operator surface over the same function, so a green test here and a red run there cannot disagree about anything except the input. ⚠ `POST /test` BLOCKS FOR ~10-30 s and COSTS A FRACTION OF A CENT. It is `def`, not `async def`, so FastAPI runs it in the threadpool and one test cannot stall the event loop for everybody. ⚠ ON THE URL IT WILL FETCH: the fetch happens inside an ephemeral HF Job on Hugging Face's network, never from this server, so this is not a door into our own infrastructure. It is still admin-gated, because it spends money and because D-51 §5's authorisation posture ("only systems the tenant is authorised to use, at their instruction") is not something an ordinary member should be able to commit the tenant to. ⛔ THIS ROUTER IS NOT MOUNTED YET. `main.py` belongs to another lane this wave, so the one `app.include_router(routes_web_agent.router)` line is a cross-fence ask — and `verify_web_agent.py` FAILS until it lands, deliberately: three finished routers once shipped 404-dead behind entirely green gates, and a gate that tolerates it is how that happens twice. """ from fastapi import APIRouter, Body, Depends import web_agent from deps import Session, require_session, err from routes_admin import admin_gate router = APIRouter(prefix="/api/v1") MAX_URL = 2000 MAX_SELECTOR = 400 @router.get("/web-agent/capability") def web_agent_capability(session: Session = Depends(require_session)): """Can a web step run here at all? Configuration, not liveness — see `web_agent.capability`. Deliberately NARROW: it answers the question a UI needs ("may I offer this, and what do I say if not") and withholds the deployment detail (namespace, which token key, the image) that only an operator has any use for. Nothing here is ever a credential. """ cap = web_agent.capability() return {"ready": bool(cap["ready"]), "reason": cap["reason"], "runnableKinds": cap["runnableKinds"], "profile": cap["profile"]} @router.post("/web-agent/test") def web_agent_test(session: Session = Depends(admin_gate), body: dict = Body(...)): """Run ONE real `web_read` and report exactly what the seam returned. The response mirrors the seam's own contract rather than flattening it: `ok` plus a `value`, or `ok:false` plus the SENTENCE. A test surface that turns a named failure into "something went wrong" would hide the one thing it exists to show. """ url = str(body.get("url") or "").strip() selector = str(body.get("selector") or "").strip() if len(url) > MAX_URL or len(selector) > MAX_SELECTOR: raise err(400, "too_long", "the URL or selector is longer than this door accepts") step = {"kind": "web_read", "id": "test", "url": url, "selector": selector, "attr": (body.get("attr") or "text"), "all": bool(body.get("all")), "timeoutMs": int(body.get("timeoutMs") or 20000)} if body.get("waitFor"): step["waitFor"] = str(body["waitFor"]) notes = [] result, error = web_agent.run_step( step, {"tenant": session.tenant, "runId": f"test-{session.tenant}-{session.uname}", "log": notes.append}) if error: # 200 with `ok:false`, not a 4xx: the request was well-formed and the ANSWER is that the # web step did not succeed. A 500 here would make an ordinary "the selector matched # nothing" look like a server fault. return {"ok": False, "error": error, "log": notes[-6:]} return {"ok": True, "result": result, "log": notes[-6:]}