api/web_agent.py

"""web_agent.py — THE SEAM between an automation step and a browser that runs somewhere else.

CONTRACT C5 (wave 31, ruling R10 / D-51). E ships this; **C mounts it** in
`automation_engine.py`'s action dispatch for `web_read` and adds the `verify_wiring` row —
⛔ E cannot verify its own mounting, and an unmounted runner is a whole, correct, UNREACHABLE
feature, which is exactly how five wave-29 features shipped behind green gates.

    result, error = web_agent.run_step(step, ctx)      # -> (dict|None, str)

⛔ **EXACTLY ONE OF THE TWO IS TRUTHY, ALWAYS.** Success is `(result, "")`; failure is
`(None, "<one sentence naming the cause>")`. Never `(None, "")` — a failure that writes nothing
and says nothing is a recorded defect class here, and it is the shape that makes an automation
loop forever writing blanks while every gate stays green. This module NEVER raises: every
exception is caught at the boundary and turned into a sentence, because the caller is a record
walk that must not die on one bad URL.

────────────────────────────────────────────────────────────────────────────────────────────────
WHY A JOB AND NOT A BROWSER HERE

Two laws, neither negotiable: the browser runs on cloud compute and never on a developer's
laptop (W27/R6 + the `/browse` law), and it lives in an EPHEMERAL job so the Space image does not
grow and €0 holds (D-51/R2). So this module submits `jobs/web_agent_job.py` to HF Jobs, polls
**past `SCHEDULING`** to a forward-progress marker the job's own script printed, reads the result
out of the log, and returns it. `aios-web/api/verify_web_agent.py` fails if any shipped module
could launch a browser in-process, this one included.

⚠ IT BLOCKS. Measured cold start is **9.3 s warm / ~24 s on a cold node** on the fast profile and
~28-32 s on the portable one (`waves/wave31/proto/web-agent-job.md` §4). That is fine for a
foreground step of an automation RUN and far too slow for a route a person is waiting on. The
cost is per JOB, not per step — `run_plan` sends a whole flow's steps in ONE job for that reason.

────────────────────────────────────────────────────────────────────────────────────────────────
⛔ THREE THINGS ARE TRUE OF PRODUCTION TODAY AND ARE NOT DEFECTS IN THIS FILE. `capability()`
reports each as a sentence, and `verify_web_agent.py` holds each to a check, so none of them can
be discovered by a customer instead of by us:

  1. THE SPACE'S TOKEN CANNOT PAY FOR A JOB. `deploy_web.py` pushes `HF_TOKEN`, which is scoped to
     the org `royal-imports`; HF answers **402 `Pre-paid credit balance is insufficient`** there
     and 0 jobs have ever run under it. `AIOS_HF_TOKEN` (user `fsanyoto`, 100 jobs) works and is
     NOT pushed. Measured 2026-08-12.
  2. THE SPACE DOES NOT SHIP THE JOB SCRIPT. `deploy_web.py` uploads `api/*.py` and
     `platform/{core,modules,harness}`; `jobs/` is in no manifest. And `check_upload()` walks
     IMPORTS, so it structurally cannot see a file opened by PATH — its own comments say so, about
     `aios_grid_fields.json`, which crashed a Space for exactly this reason.
  3. `routes_web_agent.py` IS NOT MOUNTED. `main.py` is another lane's file this wave.

Each needs one line somebody else owns. Until then this module answers with a sentence rather
than a silence, which is the whole of R6's second half.
"""
import base64
import hashlib
import json
import os
import pathlib
import time

#: ⭐ ONE PAIR OF CONSTANTS, AND THEY MUST NAME THE SAME VERSION. The fast profile runs on a
#: Microsoft image that ships the BROWSERS for one Playwright version and NOT the Python package
#: (measured — `proto/web-agent-job.md` §6), so the image TAG and the pip PIN are a single
#: decision wearing two spellings. `verify_web_agent.py` asserts they agree, and also that
#: `jobs/requirements-web-agent.txt` and the job's own PEP-723 block agree with both: a constant
#: two features share is a defect class this repo has already paid for.
PLAYWRIGHT_VERSION = "1.62.0"

#: A profile is (image, how to reach an interpreter that HAS playwright, what to install in-job).
#: Two exist for one reason: the fast one depends on a THIRD-PARTY REGISTRY, and a capability with
#: no second route dies the day that registry blinks.
#: ⚠ `pip` IS PER-PROFILE AND IT IS NOT COSMETIC. The prebuilt image carries no Python packages
#: at all, so whatever the job imports must be listed here — and the CHECKPOINT import is
#: conditional, so the list is too. This was a real defect: `huggingface_hub` was absent, the
#: checkpoint upload raised `ModuleNotFoundError`, and the run still reported a correct result
#: because the boundary caught it — i.e. the resume half was silently off while the read half
#: looked perfect. `uv` needs no list: it reads the job's own PEP-723 block.
PROFILES = {
    # 9.3 s warm to a parsed value. Browsers preinstalled; only the driver is pip-installed.
    "prebuilt": {
        "image": f"mcr.microsoft.com/playwright/python:v{PLAYWRIGHT_VERSION}-noble",
        "pip": [f"playwright=={PLAYWRIGHT_VERSION}"],
        "pip_checkpoint": ["huggingface_hub"],
        "python": "python3 -u",
        "install_deps": "0", "install_browser": "0",
    },
    # ~28-32 s. No external registry and no version coupling: uv reads the job's own PEP-723
    # block, then the job apt-installs the libraries and downloads the browser itself.
    "portable": {
        "image": "ghcr.io/astral-sh/uv:python3.12-bookworm",
        "pip": [], "pip_checkpoint": [],
        "python": "uv run",
        "install_deps": "1", "install_browser": "1",
    },
}
DEFAULT_PROFILE = "prebuilt"

#: The log protocol. `jobs/web_agent_job.py` spells these too; they are a cross-file contract
#: between two processes that never import each other, so the gate asserts both sides agree.
MARK = "[web-agent]"
BOOT_MARK = f"{MARK} BOOT"
RESULT_BEGIN = f"{MARK} RESULT-BEGIN"
RESULT_END = f"{MARK} RESULT-END"

#: ⭐⭐ ALL FIVE WEB ACTIONS RUN — owner ruling 2026-08-12, superseding D-51/R5.
#: R5 made any web action that WRITES to a third party default to "needs approval". The owner has
#: revoked it, verbatim: *"make sure we unblock all web actions, we don't need approval step
#: first wtf, I never ask for that."* There is no approval gate and no per-kind ban.
#: ⚠ The runner enforces the same set independently, so neither end has to trust the other.
RUNNABLE_KINDS = ("web_read", "web_goto", "web_fill", "web_click", "web_repair")

#: What each kind cannot run without. Checked HERE as well as in the job because a 10-second
#: round trip to be told "you gave me no selector" is a round trip nobody should pay for.
REQUIRED_FIELDS = {
    "web_goto": ("url",),
    "web_read": ("selector",),
    "web_fill": ("selector", "value"),
    "web_click": ("selector",),
    "web_repair": ("selector",),
}
#: Kinds that CHANGE something on somebody else's site. Not a wall — nothing here refuses them
#: (the owner's ruling above). It exists so the run log, the dry-run affordance and any future
#: audit surface can all ask one question in one place instead of re-listing three strings.
WRITES_TO_THIRD_PARTY = ("web_fill", "web_click")

#: ⛔ THE SCRIPT LIVES AT A DIFFERENT DEPTH IN THE REPO AND IN THE CONTAINER, so a single
#: `parents[N]` is right in exactly one of the two places this module runs. Measured at wave 31
#: close, by A, before deploy #2:
#:   repo      `<repo>/aios-web/api/web_agent.py`  -> parents[2] == `<repo>`  -> `<repo>/jobs` ✅
#:   Space     `/app/api/web_agent.py`             -> parents[2] == `/`       -> `/jobs`      ❌
#: because `aios-web/Dockerfile` flattens `api/` to `/app/api/`. The original constant was correct
#: locally and silently wrong in production — `capability()` would have reported "the browser job's
#: script is missing from this deployment": honest, and useless, which is exactly the outcome this
#: module's own header warns about. Both candidates are tried, nearest first, so neither layout is
#: privileged and a future move breaks loudly rather than quietly.
_HERE = pathlib.Path(__file__).resolve()
_CANDIDATES = (_HERE.parents[1] / "jobs",      # container: /app/api -> /app/jobs
               _HERE.parents[2] / "jobs")      # repo:      aios-web/api -> <repo>/jobs
JOB_SCRIPT = next((c / "web_agent_job.py" for c in _CANDIDATES
                   if (c / "web_agent_job.py").is_file()),
                  _CANDIDATES[-1] / "web_agent_job.py")

#: Env knobs, all optional. Names are spelled once and read through `_cfg` so the gate can find
#: them and `capability()` can report them without a second list.
ENV_TOKEN = ("AIOS_WEB_AGENT_TOKEN", "AIOS_HF_TOKEN", "HF_TOKEN")
ENV_NAMESPACE = "AIOS_WEB_AGENT_NAMESPACE"
ENV_PROFILE = "AIOS_WEB_AGENT_PROFILE"
ENV_FLAVOR = "AIOS_WEB_AGENT_FLAVOR"
ENV_REPO = "AIOS_WEB_AGENT_REPO"          # optional checkpoint store; absent is fine and stated

START_TIMEOUT = float(os.environ.get("AIOS_WEB_AGENT_START_TIMEOUT") or 180)
RUN_TIMEOUT = float(os.environ.get("AIOS_WEB_AGENT_RUN_TIMEOUT") or 300)
JOB_TIMEOUT = os.environ.get("AIOS_WEB_AGENT_JOB_TIMEOUT") or "10m"
DEFAULT_FLAVOR = "cpu-basic"              # measured no slower than cpu-upgrade, and 3x cheaper
POLL_SECONDS = 2.0
MAX_STEPS = 20                            # matches the runner's own ceiling and MAX_ACTIONS

_ns_cache = {}


def _token():
    for key in ENV_TOKEN:
        val = (os.environ.get(key) or "").strip()
        if val:
            return val, key
    return "", ""


def _profile():
    name = (os.environ.get(ENV_PROFILE) or DEFAULT_PROFILE).strip()
    return (name if name in PROFILES else DEFAULT_PROFILE), PROFILES.get(name) or \
        PROFILES[DEFAULT_PROFILE]


def _api(token):
    from huggingface_hub import HfApi          # already a dependency: the store is HF-backed
    return HfApi(token=token)


def _namespace(token):
    """Which HF namespace the job is billed to. Explicit env wins; otherwise the token's own
    owner, resolved ONCE per process — `whoami` is rate-limited and this answer never changes."""
    fixed = (os.environ.get(ENV_NAMESPACE) or "").strip()
    if fixed:
        return fixed, ""
    if token in _ns_cache:
        return _ns_cache[token], ""
    try:
        name = str(_api(token).whoami().get("name") or "")
    except Exception as exc:                    # noqa: BLE001 — a boundary
        return "", (f"Hugging Face would not say who this deployment's token belongs to "
                    f"({type(exc).__name__}), so there is no namespace to bill a browser job to.")
    if not name:
        return "", "Hugging Face returned no account name for this deployment's token."
    _ns_cache[token] = name
    return name, ""


def _status_of(exc):
    """The HTTP status of an HF error, from the RESPONSE — never by grepping the message.

    ⛔ THIS WAS A REAL DEFECT AND IT LIED IN THE MOST EXPENSIVE DIRECTION. The first version asked
    `if "402" in text`, and an `HfHubHTTPError` message carries the URL and a random hex Request
    ID — so a genuine **403 "Make sure your token has the correct permissions"** was reported to
    the operator as *"the account has no prepaid Jobs credit"*, sending them to buy credit for a
    problem credit cannot fix. Measured 2026-08-12, both directions, on live HF. A status code is
    a NUMBER the response already carries; matching it as a substring of free text is matching it
    against every request id in the world. [[one-question-two-normalizers]]
    """
    resp = getattr(exc, "response", None)
    code = getattr(resp, "status_code", None)
    return int(code) if isinstance(code, int) else None


def _sentence_for(exc, namespace):
    """Turn an HF exception into ONE sentence a person can act on.

    402 and 403 get their own arms because neither is a bug in our code and both read like one —
    and they need OPPOSITE fixes, which is exactly why they must not be confused: 402 is money on
    an account we are allowed to bill, 403 is a token that may not bill this account at all.
    """
    status, text = _status_of(exc), f"{exc}"
    where = namespace or "the configured account"
    if status == 402:
        return (f"Hugging Face refused the browser job: the account it would be billed to "
                f"({where}) has no prepaid Jobs credit. Either add credit to that account, or "
                f"point this deployment at one that has it ({ENV_NAMESPACE} / {ENV_TOKEN[0]}). "
                f"Nothing was read.")
    if status in (401, 403):
        return (f"Hugging Face refused the browser job as unauthorised for {where} — this "
                f"deployment's token does not carry `job.write` on that account. Point it at an "
                f"account the token owns ({ENV_NAMESPACE}), or configure a token that owns this "
                f"one ({ENV_TOKEN[0]}). Nothing was read.")
    return (f"The browser job could not be submitted "
            f"({type(exc).__name__}{f' {status}' if status else ''}: "
            f"{text.splitlines()[0][:200]}). Nothing was read.")


def capability():
    """Can this deployment run a web step AT ALL, and if not, WHY — WITHOUT calling HF.

    This exists so the answer is visible BEFORE a customer's automation fails at 3am. It reports
    configuration, not liveness: a green `ready` means nothing is missing here, not that HF will
    accept the next submission (only `run_step` can learn that, and it reports it as a sentence).
    """
    token, token_key = _token()
    name, prof = _profile()
    reasons = []
    if not token:
        reasons.append(f"no Hugging Face token is configured ({' / '.join(ENV_TOKEN)})")
    if not JOB_SCRIPT.is_file():
        reasons.append(f"the browser job's script is missing from this deployment "
                       f"(expected {JOB_SCRIPT.name} under jobs/)")
    return {
        "ready": not reasons,
        "reason": ("; ".join(reasons) or "configured"),
        "tokenKey": token_key,
        "namespace": (os.environ.get(ENV_NAMESPACE) or "").strip() or "(the token's own account)",
        "profile": name,
        "image": prof["image"],
        "flavor": (os.environ.get(ENV_FLAVOR) or DEFAULT_FLAVOR),
        "scriptPresent": JOB_SCRIPT.is_file(),
        "checkpointRepo": (os.environ.get(ENV_REPO) or "").strip(),
        "runnableKinds": list(RUNNABLE_KINDS),
        "playwrightVersion": PLAYWRIGHT_VERSION,
    }


def _clean_step(step, index, first=True):
    """A step as the runner will see it, or a sentence. Validated HERE as well as in the job:
    a 10-second round trip to be told a URL is empty is a round trip nobody should pay for.

    `first` says whether this step opens the journey. Only the FIRST step must carry a `url` —
    every later one may continue on the page its predecessor left behind, which is the whole
    point of a journey (open, fill, click, read).
    """
    kind = str((step or {}).get("kind") or "")
    if kind not in RUNNABLE_KINDS:
        return None, (f"This step asks for `{kind or '(nothing)'}`, which is not a web action. "
                      f"The ones that run are {', '.join(RUNNABLE_KINDS)}.")

    url = str(step.get("url") or "").strip()
    if url and not url.lower().startswith(("http://", "https://")):
        return None, (f"Step {index} has a web address that is not http or https ({url!r}), so "
                      f"there is nothing to open.")
    if first and not url:
        return None, (f"Step {index} is the first step of this journey, so it needs a web "
                      f"address to open. Later steps continue on the same page.")

    for field in REQUIRED_FIELDS.get(kind, ()):
        if field == "url":
            continue                            # handled above, with a better sentence
        if field == "value":
            if step.get("value") is None:
                return None, (f"Step {index} fills a field but carries no value to type into "
                              f"it.")
            continue
        if not str(step.get(field) or "").strip():
            return None, (f"Step {index} ({kind}) names no element ({field}). Give it a CSS "
                          f"selector, such as `#username` or `button[type=submit]`.")

    out = {"id": str(step.get("id") or f"s{index}"), "kind": kind,
           "timeoutMs": int(step.get("timeoutMs") or 20000)}
    if url:
        out["url"] = url
    for key in ("selector", "attr", "waitFor", "hint"):
        if step.get(key):
            out[key] = str(step[key])
    if kind == "web_fill":
        out["value"] = str(step.get("value"))
        # An explicit `secret` wins; otherwise the runner infers it from the field name. Carried
        # through so the decision is made once, by whoever knows, and not re-guessed downstream.
        if step.get("secret") is not None:
            out["secret"] = bool(step["secret"])
    if step.get("all"):
        out["all"] = True
    if step.get("dryRun"):
        out["dryRun"] = True
    return out, ""


def _shell_command(plan, prof):
    """The job's whole command: the script, base64-embedded, then the profile's runner.

    ⛔ EMBEDDED RATHER THAN UPLOADED, DELIBERATELY. `run_uv_job`'s local-file path uploads to an
    HF bucket and mounts it, which adds a storage dependency and a second thing to go stale; the
    base64 route is self-contained, costs no repo commit, and cannot ship a version of the script
    that differs from the one on disk.
    """
    src = JOB_SCRIPT.read_bytes()
    b64 = base64.b64encode(src).decode("ascii")
    plan_b64 = base64.b64encode(json.dumps(plan).encode("utf-8")).decode("ascii")
    repo_configured = bool((os.environ.get(ENV_REPO) or "").strip())
    pkgs = list(prof["pip"]) + (list(prof["pip_checkpoint"]) if repo_configured else [])
    prelude = (f"pip install --break-system-packages -q {' '.join(pkgs)} && " if pkgs else "")
    shell = (f"printf '%s' '{b64}' | base64 -d > /tmp/web_agent_job.py && "
             f"{prelude}{prof['python']} /tmp/web_agent_job.py")
    env = {"WEB_AGENT_PLAN": plan_b64,
           "WEB_AGENT_INSTALL_DEPS": prof["install_deps"],
           "WEB_AGENT_INSTALL_BROWSER": prof["install_browser"]}
    #: ⭐ THE JOB GETS A CREDENTIAL ONLY WHEN IT HAS SOMEWHERE TO WRITE. With no checkpoint repo
    #: configured, no token is handed to the browser container at all — which is the better
    #: posture and the default. `secrets` takes actual VALUES (never a key name, never an array:
    #: `run_uv_job`'s array form 400s), and HF delivers them as env inside the container.
    secrets = {}
    repo = (os.environ.get(ENV_REPO) or "").strip()
    if repo:
        env["WEB_AGENT_REPO"] = repo
        token, _ = _token()
        if token:
            secrets["HF_TOKEN"] = token
    return ["/bin/sh", "-c", shell], env, secrets


def run_plan(steps, ctx=None):
    """Run a whole flow's web steps in ONE job. `run_step` is the one-step case.

    Returns `(results, "")` where results is a list aligned to `steps`, or `(None, sentence)`.
    """
    ctx = ctx or {}
    log = ctx.get("log") or (lambda _m: None)

    if not steps:
        return None, "No web step was given, so nothing was read."
    if len(steps) > MAX_STEPS:
        return None, (f"A single browser job carries at most {MAX_STEPS} web steps and this flow "
                      f"has {len(steps)}. Split the flow.")

    cleaned = []
    for idx, step in enumerate(steps, 1):
        one, why = _clean_step(step, idx, first=(idx == 1))
        if why:
            return None, why
        cleaned.append(one)

    cap = capability()
    if not cap["ready"]:
        return None, (f"This deployment cannot run a web step: {cap['reason']}. Nothing was read.")

    token, _ = _token()
    namespace, why = _namespace(token)
    if why:
        return None, why
    prof_name, prof = _profile()
    #: ⛔⛔ THE CHECKPOINT KEY IS (caller's runId + THE WORK), NEVER THE runId ALONE, and getting
    #: this wrong returns one page's content for a different page's request — silently, and only
    #: once a checkpoint store is configured. The job resumes on a runId hit by returning the
    #: prior envelope UNCHANGED, so any caller that reuses an id across different work gets the
    #: first answer forever. Two real callers would have done exactly that: this module's own
    #: `/web-agent/test` route keys on `test-<tenant>-<user>` (constant per person — test URL A,
    #: then URL B, get A back), and the automation engine walks records with ONE run id, so
    #: records 2..N would all receive record 1's scraped value. Hashing the cleaned steps in makes
    #: "resume" mean *this exact work already completed*, which is the only thing it may mean.
    #: ⚠ `cleaned` is used, not the caller's raw dict: it is normalised and ordered, so two
    #: spellings of one step share a key and a genuine difference cannot hide in a default.
    run_id = str(ctx.get("runId") or "").strip() or f"auto-{int(time.time())}"
    work = hashlib.sha256(
        json.dumps(cleaned, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
    plan = {"runId": f"{run_id}-{work[:12]}", "steps": cleaned, "callerRunId": run_id}

    try:
        command, env, secrets = _shell_command(plan, prof)
    except OSError as exc:
        return None, (f"The browser job's script could not be read from this deployment "
                      f"({type(exc).__name__}). Nothing was read.")

    api = _api(token)
    started = time.monotonic()
    try:
        job = api.run_job(image=prof["image"], command=command, env=env, secrets=secrets,
                          flavor=(os.environ.get(ENV_FLAVOR) or DEFAULT_FLAVOR),
                          timeout=JOB_TIMEOUT, namespace=namespace,
                          labels={"aios": "web-agent", "profile": prof_name})
    except Exception as exc:                     # noqa: BLE001 — a boundary
        return None, _sentence_for(exc, namespace)

    job_id = str(getattr(job, "id", "") or "")
    log(f"[web-agent] submitted {job_id} ({prof_name}, {namespace}) for {len(cleaned)} step(s)")

    result, why = _await(api, job_id, namespace, started, log)
    if why:
        return None, why
    rows = result.get("steps", [])
    for row in rows:
        row["jobId"] = job_id

    # ⛔ A JOURNEY THAT STOPPED HALFWAY IS A FAILURE, and returning its rows with an empty error
    # is how a caller "succeeds" on a login that never happened. Measured: a 3-step plan whose
    # click matched nothing came back `(rows, "")` because the envelope carried per-step errors
    # and no top-level one — every row was inspectable and nothing forced anybody to inspect
    # them. The contract this module publishes is exactly-one-truthy, so it has to hold for a
    # plan too. The sentence names the failing step AND what completed before it, because that
    # is the whole diagnostic value the rows would have carried.
    failed = [r for r in rows if not r.get("ok") and r.get("attempted") is not False]
    if failed:
        first = failed[0]
        reason = str(first.get("error") or "no reason given")
        # ⚠ ONE step is not a "journey", and wrapping its sentence in journey prose would make
        # the single-step case — the one the automation engine actually calls — read worse than
        # before. The wrapper earns its place only when there were other steps to lose.
        if len(rows) == 1:
            return None, reason
        done = [str(r.get("id") or r.get("kind")) for r in rows if r.get("ok")]
        skipped = [r for r in rows if r.get("attempted") is False]
        return None, (f"Step {rows.index(first) + 1} of {len(rows)} ({first.get('kind')}) failed "
                      f"and the journey stopped: {reason}"
                      + (f" Completed first: {', '.join(done)}." if done else "")
                      + (f" Not attempted: {len(skipped)} later step(s)." if skipped else ""))
    return rows, ""


def run_step(step, ctx=None):
    """C5's seam: ONE declarative web step -> `(result, error_sentence)`.

    `step`  {"kind": "web_read", "url", "selector", "attr"?, "all"?, "waitFor"?, "timeoutMs"?}
    `ctx`   optional {"tenant", "automationId", "runId", "log"}
    """
    try:
        rows, why = run_plan([step], ctx)
    except Exception as exc:                     # noqa: BLE001 — the LAST boundary
        # Nothing below may reach the caller as an exception: this runs inside a record walk.
        return None, (f"The web step failed unexpectedly ({type(exc).__name__}: "
                      f"{str(exc).splitlines()[0][:200]}). Nothing was read.")
    if why:
        return None, why
    if not rows:
        # Belt and braces for the contract's own edge: a job that returned ok with no rows.
        return None, ("The browser job finished without returning anything for this step. "
                      "Nothing was read.")
    row = rows[0]
    if not row.get("ok"):
        return None, str(row.get("error") or
                         "The web step did not succeed and the job gave no reason.")
    return row, ""


def _await(api, job_id, namespace, started, log):
    """Poll past `SCHEDULING`, tail to the job's OWN marker, return its result envelope.

    ⛔ NEVER 'RUNNING' ON SUBMISSION ALONE — the HF-jobs rule, and the reason this function is
    longer than a sleep. Two deadlines, because the two failures are different and a person needs
    to be told which one happened: a job that never STARTS is a platform queue problem, and a job
    that starts and never SPEAKS is ours.
    """
    seen, state, booted = 0, "", False
    envelope = None

    while True:
        waited = time.monotonic() - started
        try:
            info = api.inspect_job(job_id=job_id, namespace=namespace)
            state = str(getattr(info.status, "stage", info.status) or "")
        except Exception as exc:                 # noqa: BLE001
            return None, (f"The browser job {job_id} was submitted but its status could not be "
                          f"read ({type(exc).__name__}). Nothing was read.")

        if state in ("RUNNING", "COMPLETED", "ERROR", "CANCELED"):
            # ⚠ `fetch_job_logs` REPLAYS THE WHOLE LOG on every call, so the index is the dedupe.
            # A content-based one would swallow legitimately repeated lines.
            try:
                lines = list(api.fetch_job_logs(job_id=job_id, namespace=namespace))
            except Exception:                    # noqa: BLE001 — logs lag a fresh container
                lines = []
            fresh, seen = lines[seen:], max(seen, len(lines))
            buf, capture = [], False
            for raw in fresh:
                line = (raw or "").rstrip()
                if not line:
                    continue
                if not booted and BOOT_MARK in line:
                    booted = True
                    log(f"[web-agent] job {job_id} is alive at "
                        f"+{time.monotonic() - started:.1f}s")
                if RESULT_BEGIN in line:
                    capture, buf = True, []
                    continue
                if RESULT_END in line:
                    capture = False
                    try:
                        envelope = json.loads("".join(buf))
                    except Exception:            # noqa: BLE001
                        envelope = None
                    continue
                if capture:
                    buf.append(line)

        if envelope is not None:
            break
        if state in ("COMPLETED", "ERROR", "CANCELED"):
            return None, (f"The browser job {job_id} ended {state} without returning a result"
                          f"{' — it never started printing' if not booted else ''}. "
                          f"Nothing was read.")
        if not booted and waited > START_TIMEOUT:
            # The submitted-but-never-started case, named as itself.
            try:
                api.cancel_job(job_id=job_id, namespace=namespace)
            except Exception:                    # noqa: BLE001 — best effort; it may be gone
                pass
            return None, (f"The browser job {job_id} was accepted but had not started after "
                          f"{int(START_TIMEOUT)}s (it stayed {state or 'queued'}), so it was "
                          f"cancelled. Nothing was read — try again, or the platform is busy.")
        if waited > RUN_TIMEOUT:
            try:
                api.cancel_job(job_id=job_id, namespace=namespace)
            except Exception:                    # noqa: BLE001
                pass
            return None, (f"The browser job {job_id} ran for more than {int(RUN_TIMEOUT)}s "
                          f"without finishing and was cancelled. The page may be too slow or may "
                          f"never load. Nothing was read.")
        time.sleep(POLL_SECONDS)

    if not envelope.get("ok") and envelope.get("error"):
        # The runner's OWN sentence, passed through untouched. Rewriting it here would put two
        # vocabularies on one failure.
        return None, str(envelope["error"])
    return envelope, ""