web_tools.py

"""Live web-tools verification harness (NOT a pytest module).

Drives a real, billable deploy + query of the focused ``web-tools`` fixture against
Google Vertex AI Agent Engine and records an objective, re-runnable receipt. The
fixture exercises agentlift's built-in web mapping in one folder:

  web_search  -> Gemini Google Search grounding, lowered as a dedicated single-tool
                 ADK sub-agent wrapped in an AgentTool (propagate_grounding_metadata)
  web_fetch   -> Gemini URL Context, lowered the same way (approximate)

Three agents: a ``lead`` coordinator (own web_search + sub_agents), a search-only
``searcher`` leaf, and a ``fetcher`` carrying BOTH web tools. The always-wrap design
means the coordinator's web_search never collides with the injected transfer tools.

Per the project's live discipline, every dimension is reported in one of four states:

  PASS-WIRED     deploy/package proves the feature was configured (deterministic)
  PASS-EXERCISED runtime stream proves the provider actually used it (objective event)
  NOT-PROVEN     wired correctly, but no objective runtime signal (model chose not to)
  FAIL           deploy / config / runtime error or wrong behavior

Objective runtime signals (asserted on the event stream, never on answer text):
  - web_search : grounding_metadata.web_search_queries nonempty OR grounding_chunks
                 present (the wrapped google_search actually executed and grounded)
  - web_fetch  : url_context_metadata.url_metadata[*].retrieved_url present (the
                 wrapped url_context actually fetched a page)
  - delegation : a transfer_to_agent function-call (coordinator -> searcher/fetcher)
  - wrapper    : an AgentTool function-call named <agent>_web_search / _web_fetch
                 (corroborates that the lowered tool-agent was invoked)

Usage (each billable step is explicit; deploy is slow, queries are cheap):
  python tests/live/web_tools.py deploy-google      # ~minutes, billable
  python tests/live/web_tools.py query-google        # cheap; writes receipt
  python tests/live/web_tools.py teardown-google

Credentials (never committed): ADC + GOOGLE_CLOUD_PROJECT + GOOGLE_CLOUD_LOCATION +
AGENTLIFT_GCP_STAGING_BUCKET. Receipts land under tests/live/receipts/ and are
committed as auditable evidence (Google project/resource values anonymized with
``****`` before commit). This file is NOT collected by pytest.
"""
from __future__ import annotations

import datetime
import json
import os
import sys
import traceback
from typing import Any

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
SRC = os.path.join(ROOT, "src")
if SRC not in sys.path:
    sys.path.insert(0, SRC)

FIXTURE = os.path.join(HERE, "fixtures", "web-tools")
RECEIPTS = os.path.join(HERE, "receipts")
RUN_USER = "agentlift-live"

# Airtight URL-fetch canary: httpbingo's /base64/<b64> endpoint echoes the decoded
# token as the page body. The token is URL-derived and NOT in any training corpus, so
# a model can only produce it by actually retrieving the page (example.com's text is
# memorizable; this isn't). Stable across runs -> doubles as a regression canary.
CANARY_NONCE = "AGENTLIFT-URLCTX-9F3A2C7E-CANARY"
CANARY_URL = "https://httpbingo.org/base64/QUdFTlRMSUZULVVSTENUWC05RjNBMkM3RS1DQU5BUlk="

# Queries are split so one query never conflates web_search with web_fetch.
Q_SEARCH = (
    "What is the Agent Engine in Google Vertex AI, in one or two sentences? "
    "Search the web for a current answer and include the source URL you relied on. "
    "Do not answer from memory."
)
Q_FETCH = (
    f"Fetch the web page at {CANARY_URL} and tell me, verbatim, the exact text shown "
    "on that page. Quote it exactly and cite the URL you retrieved. Use a URL-retrieval "
    "tool; do not answer from memory."
)


def _ts() -> str:
    return datetime.datetime.now().strftime("%Y%m%d-%H%M%S")


def _ensure(d: str) -> str:
    os.makedirs(d, exist_ok=True)
    return d


def _versions() -> dict[str, str]:
    out: dict[str, str] = {}
    for mod in ("google.adk", "vertexai", "agentlift"):
        try:
            m = __import__(mod, fromlist=["__version__"])
            out[mod] = getattr(m, "__version__", "?")
        except Exception as e:
            out[mod] = f"<{type(e).__name__}>"
    return out


# --------------------------------------------------------------------------- #
# event extraction (robust to snake_case / camelCase serialization)
# --------------------------------------------------------------------------- #
def _walk(obj: Any):
    if isinstance(obj, dict):
        yield obj
        for v in obj.values():
            yield from _walk(v)
    elif isinstance(obj, list):
        for v in obj:
            yield from _walk(v)


def _norm(k: str) -> str:
    return k.replace("_", "").lower()


def google_event_calls(ev: dict) -> list[dict[str, Any]]:
    calls = []
    for d in _walk(ev):
        if not isinstance(d, dict):
            continue
        fc = d.get("function_call") or d.get("functionCall")
        if isinstance(fc, dict) and fc.get("name"):
            calls.append({"name": fc["name"], "args": fc.get("args") or {}})
    return calls


def google_event_responses(ev: dict) -> list[dict[str, Any]]:
    """Structured tool outputs (function_response) on the stream -- objective receipt
    material, distinct from the model's free-form answer text."""
    out = []
    for d in _walk(ev):
        if not isinstance(d, dict):
            continue
        fr = d.get("function_response") or d.get("functionResponse")
        if isinstance(fr, dict) and fr.get("name"):
            resp = fr.get("response") or {}
            result = resp.get("result") if isinstance(resp, dict) else resp
            out.append({"name": fr["name"], "result": "" if result is None else str(result)})
    return out


def google_event_text(ev: dict) -> str:
    chunks = []
    for d in _walk(ev):
        if isinstance(d, dict) and isinstance(d.get("text"), str):
            chunks.append(d["text"])
    return "".join(chunks)


def harvest_metadata(ev: dict) -> dict[str, Any]:
    """Pull grounding + url_context signals out of one event, tolerant of the
    snake_case (pydantic attr) vs camelCase (genai proto) split."""
    queries: list[str] = []
    chunks = 0
    retrieved: list[dict[str, Any]] = []
    for d in _walk(ev):
        if not isinstance(d, dict):
            continue
        for k, v in d.items():
            nk = _norm(k)
            if nk == "websearchqueries" and isinstance(v, list):
                queries.extend(str(q) for q in v)
            elif nk == "groundingchunks" and isinstance(v, list):
                chunks += len(v)
            elif nk == "retrievedurl" and v:
                # sibling key carries the retrieval status when present
                status = d.get("url_retrieval_status") or d.get("urlRetrievalStatus")
                retrieved.append({"url": v, "status": status})
    return {"web_search_queries": queries, "grounding_chunks": chunks,
            "retrieved_urls": retrieved}


# --------------------------------------------------------------------------- #
# Google: deploy / query / teardown
# --------------------------------------------------------------------------- #
def _google_env() -> tuple[str, str, str]:
    proj = os.environ.get("GOOGLE_CLOUD_PROJECT")
    bucket = os.environ.get("AGENTLIFT_GCP_STAGING_BUCKET")
    loc = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
    if not proj or not bucket:
        raise SystemExit("set GOOGLE_CLOUD_PROJECT and AGENTLIFT_GCP_STAGING_BUCKET (gs://...)")
    return proj, loc, bucket


def deploy_google() -> dict[str, Any]:
    import warnings
    warnings.filterwarnings("ignore")
    os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "TRUE"
    from agentlift.parser import parse_project
    from agentlift.google_plan import build_google_plan
    from agentlift.google_target import deploy_google as _deploy

    proj, loc, bucket = _google_env()
    project, _diags = parse_project(FIXTURE)
    plan = build_google_plan(project)
    if not plan.deployable:
        raise SystemExit("google plan not deployable:\n" + plan.diagnostics.render())

    print(f"deploying web-tools to Agent Engine (project={proj} region={loc})...")
    t0 = datetime.datetime.now()
    res = _deploy(project, gcp_project=proj, location=loc, staging_bucket=bucket, log=print)
    dur = (datetime.datetime.now() - t0).total_seconds()

    state = {
        "provider": "google", "fixture": "web-tools", "when": _ts(),
        "deploy_seconds": round(dur, 1), "action": res.action,
        "resource_name": res.resource_name, "spec_hash": res.spec_hash,
        "display_name": res.display_name, "deploy_model": res.deploy_model,
        "project": proj, "location": loc, "plan": plan.to_dict(),
        "versions": _versions(),
    }
    path = os.path.join(_ensure(RECEIPTS), "_state-web-google.json")
    json.dump(state, open(path, "w", encoding="utf-8"), indent=2)
    print(f"\n{res.action}: {res.resource_name}\nwrote {path}")
    return state


def _get_google_engine(resource_name: str):
    import warnings
    warnings.filterwarnings("ignore")
    import vertexai
    from vertexai import agent_engines
    proj, loc, bucket = _google_env()
    os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "TRUE"
    vertexai.init(project=proj, location=loc, staging_bucket=bucket)
    return agent_engines.get(resource_name)


def query_google_once(engine, prompt: str) -> dict[str, Any]:
    sess = engine.create_session(user_id=RUN_USER)
    sid = sess["id"] if isinstance(sess, dict) else getattr(sess, "id", None)
    events: list[dict] = []
    calls: list[dict] = []
    responses: list[dict] = []
    text_parts: list[str] = []
    queries: list[str] = []
    chunks = 0
    retrieved: list[dict] = []
    err = None
    try:
        for ev in engine.stream_query(message=prompt, user_id=RUN_USER, session_id=sid):
            ev = ev if isinstance(ev, dict) else json.loads(json.dumps(ev, default=str))
            events.append(ev)
            calls.extend(google_event_calls(ev))
            responses.extend(google_event_responses(ev))
            t = google_event_text(ev)
            if t:
                text_parts.append(t)
            md = harvest_metadata(ev)
            queries.extend(md["web_search_queries"])
            chunks += md["grounding_chunks"]
            retrieved.extend(md["retrieved_urls"])
    except Exception as e:
        err = f"{type(e).__name__}: {e}"
    return {
        "prompt": prompt, "session_id": sid, "error": err,
        "events": events, "tool_calls": calls, "tool_responses": responses,
        "web_search_queries": queries, "grounding_chunks": chunks,
        "retrieved_urls": retrieved,
        "final_text": "".join(text_parts).strip(),
    }


def query_google() -> dict[str, Any]:
    state = json.load(open(os.path.join(RECEIPTS, "_state-web-google.json"), encoding="utf-8"))
    engine = _get_google_engine(state["resource_name"])
    print(f"querying {state['resource_name']}")
    runs = []
    for label, q in [("web-search", Q_SEARCH), ("web-fetch", Q_FETCH)]:
        print(f"  query[{label}]...")
        r = query_google_once(engine, q)
        r["label"] = label
        names = sorted({c["name"] for c in r["tool_calls"]})
        print(f"    tool_calls: {names or '(none)'}")
        print(f"    grounding : queries={r['web_search_queries']} chunks={r['grounding_chunks']}")
        print(f"    retrieved : {r['retrieved_urls']}  err={r['error']}")
        runs.append(r)
    matrix = classify_google(runs)
    return _write_receipt("web-google", state, runs, matrix)


def teardown_google() -> None:
    state_path = os.path.join(RECEIPTS, "_state-web-google.json")
    if not os.path.isfile(state_path):
        print("no web-google state; nothing to tear down")
        return
    state = json.load(open(state_path, encoding="utf-8"))
    engine = _get_google_engine(state["resource_name"])
    print(f"deleting {state['resource_name']} ...")
    engine.delete(force=True)
    print("deleted.")


# --------------------------------------------------------------------------- #
# classification -> 4-state matrix
# --------------------------------------------------------------------------- #
def _all_calls(runs) -> set[str]:
    return {c["name"] for r in runs for c in r["tool_calls"]}


def classify_google(runs) -> dict[str, Any]:
    """Classify against OBJECTIVE event-stream artifacts: the AgentTool function_call
    for a lowered web tool, and its structured function_response (NOT the model's
    free-form answer text). Per the Codex consult, agentlift's observable unit is the
    *compiled* tool (the wrapped AgentTool), so a function_call(<agent>_web_search /
    _web_fetch) + a nontrivial function_response is valid PASS-EXERCISED material.

    Separately recorded (honesty): the inner google_search/url_context grounding
    metadata does NOT cross the AgentTool -> Agent Engine stream_query boundary, even
    with propagate_grounding_metadata=True -- so the structured citation surface is not
    available downstream. We note that without downgrading the runtime proof."""
    calls = _all_calls(runs)
    any_err = any(r["error"] for r in runs)
    answered = any(r["final_text"] and not r["error"] for r in runs)

    transfers = {c["args"].get("agent_name") for r in runs for c in r["tool_calls"]
                 if c["name"] == "transfer_to_agent"}

    # the lowered tool-agents, by wrapper-name suffix
    search_calls = [c for r in runs for c in r["tool_calls"] if c["name"].endswith("_web_search")]
    fetch_calls = [c for r in runs for c in r["tool_calls"] if c["name"].endswith("_web_fetch")]
    search_resps = [x for r in runs for x in r["tool_responses"] if x["name"].endswith("_web_search")]
    fetch_resps = [x for r in runs for x in r["tool_responses"] if x["name"].endswith("_web_fetch")]

    # the search-query strings the wrapped agent emitted ARE google_search behavior
    search_query_args = [c["args"].get("request") for c in search_calls if c["args"].get("request")]
    search_grounded_text = any(len((x.get("result") or "")) > 80 for x in search_resps)

    # airtight fetch proof: the URL-derived nonce can only appear if url_context fetched it
    nonce_in_resp = any(CANARY_NONCE in (x.get("result") or "") for x in fetch_resps)

    # the structured grounding/url_context surface (absent on Agent Engine stream)
    md_queries = [q for r in runs for q in r["web_search_queries"]]
    md_chunks = sum(r["grounding_chunks"] for r in runs)
    md_urls = [u for r in runs for u in r["retrieved_urls"]]
    metadata_surfaced = bool(md_queries) or md_chunks > 0 or bool(md_urls)
    md_note = ("surfaced" if metadata_surfaced
               else "not_exposed_by_agent_engine_stream")

    m: dict[str, Any] = {}
    m["agents"] = {"state": "PASS-EXERCISED" if answered else "FAIL",
                   "reason": "" if answered else "engine returned no text / errored"}

    m["delegation"] = {
        "state": "PASS-EXERCISED" if transfers else "NOT-PROVEN",
        "reason": "" if transfers else "no transfer_to_agent event (lead may have answered itself)",
        "evidence": {"transfers_to": sorted(t for t in transfers if t)},
    }

    # web_search: wrapper invoked + emitted real search queries + nontrivial grounded
    # response. Softer than metadata proof (no citation chunks), but objective.
    if search_calls and (search_query_args or search_grounded_text):
        m["web_search"] = {"state": "PASS-EXERCISED", "reason": ""}
    elif search_calls:
        m["web_search"] = {"state": "NOT-PROVEN",
                           "reason": "wrapper invoked but response was empty/trivial"}
    else:
        m["web_search"] = {"state": "PASS-WIRED",
                           "reason": "wrapped in package; model did not invoke the web_search tool-agent"}
    m["web_search"]["evidence"] = {
        "signal": "wrapped_agent_tool_call_and_response",
        "wrapper_calls": sorted({c["name"] for c in search_calls}),
        "search_query_args": search_query_args,
        "response_chars": [len(x.get("result") or "") for x in search_resps],
        "metadata": md_note,
    }

    # web_fetch: the URL-derived nonce in the wrapper's response is conclusive proof
    # url_context actually fetched the page (the token is not in any training corpus).
    if fetch_calls and nonce_in_resp:
        m["web_fetch"] = {"state": "PASS-EXERCISED", "reason": ""}
    elif fetch_calls:
        m["web_fetch"] = {"state": "NOT-PROVEN",
                          "reason": "wrapper invoked but canary nonce not present in the response"}
    else:
        m["web_fetch"] = {"state": "PASS-WIRED",
                          "reason": "wrapped in package; model did not invoke the web_fetch tool-agent"}
    m["web_fetch"]["evidence"] = {
        "signal": "wrapped_agent_tool_call_and_response",
        "wrapper_calls": sorted({c["name"] for c in fetch_calls}),
        "canary_nonce": CANARY_NONCE, "nonce_in_response": nonce_in_resp,
        "metadata": md_note,
    }

    return {"calls_seen": sorted(calls), "any_error": any_err,
            "grounding_metadata": md_note, "dimensions": m}


# --------------------------------------------------------------------------- #
# receipt
# --------------------------------------------------------------------------- #
def _write_receipt(provider: str, state: dict, runs: list, matrix: dict) -> dict[str, Any]:
    stamp = _ts()
    rdir = _ensure(os.path.join(RECEIPTS, f"{stamp}-{provider}"))
    with open(os.path.join(rdir, "events.jsonl"), "w", encoding="utf-8") as fh:
        for r in runs:
            for ev in r.get("events", []):
                fh.write(json.dumps({"label": r["label"], "event": ev}, default=str) + "\n")
    slim_runs = [{k: v for k, v in r.items() if k != "events"} for r in runs]
    receipt = {
        "provider": provider, "when": stamp, "fixture": "web-tools",
        "state": state, "queries": slim_runs, "matrix": matrix,
        "note": "Live, billable. Not run in CI (credentials are not shared). "
                "States: PASS-EXERCISED (objective runtime event), PASS-WIRED "
                "(configured + deployed, no runtime event), NOT-PROVEN, FAIL.",
    }
    path = os.path.join(rdir, "receipt.json")
    json.dump(receipt, open(path, "w", encoding="utf-8"), indent=2, default=str)
    print(f"\n=== {provider} matrix ===")
    for dim, v in matrix["dimensions"].items():
        print(f"  {dim:14s} {v['state']:15s} {v.get('reason','')}")
    print(f"\nwrote {path}")
    return receipt


# --------------------------------------------------------------------------- #
COMMANDS = {
    "deploy-google": deploy_google,
    "query-google": query_google,
    "teardown-google": teardown_google,
}


def main(argv: list[str]) -> int:
    if len(argv) != 1 or argv[0] not in COMMANDS:
        print("usage: web_tools.py {" + " | ".join(COMMANDS) + "}")
        return 2
    try:
        from agentlift.cli import load_env
        load_env(os.getcwd(), ROOT)
    except Exception:
        pass
    try:
        COMMANDS[argv[0]]()
        return 0
    except SystemExit:
        raise
    except Exception:
        traceback.print_exc()
        return 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))