feat(verify): VAS-2 — gecko verify-docs (check docs claims against reality)

gecko/cli.py

@@ -38,7 +38,7 @@
 from .access import public_session, stub_session
 from .client import AgentApiClient
 from .ingest import load_spec
-from .modes import coerce_mode
+from .modes import CallMode, coerce_mode
 from .netguard import UnsafeUrlError, validate_public_url
 
 _SUBCOMMANDS = (
@@ -50,6 +50,7 @@
     "test",
     "inspect",
     "report",
+    "verify-docs",
     "from-docs",
     "scan-image",
     "scan-doc",
@@ -353,6 +354,50 @@ def _cmd_report(argv: list[str]) -> int:
     return 0
 
 
+def _cmd_verify_docs(argv: list[str]) -> int:
+    """`gecko verify-docs <spec> [--live]` — check ops against reality (VAS-2).
+
+    Thin transport over :mod:`gecko.verify`: build the surface, verify every op, print the
+    control-plane JSON report ({op_id: {status, basis, provenance}} + counts). Default
+    recorded (offline, $0 — every op honestly UNVERIFIED); ``--live`` opts into real calls
+    (2xx VERIFIES, a 404 on a doc-claimed endpoint REFUTES).
+    """
+    from . import verify as verify_mod
+
+    p = argparse.ArgumentParser(
+        prog="gecko verify-docs",
+        description="Verify a surface's operations against the real API and report verdicts.",
+    )
+    p.add_argument(
+        "spec",
+        help="An API domain, OpenAPI URL, docs URL, or local path — Gecko finds the spec.",
+    )
+    p.add_argument(
+        "--live",
+        action="store_true",
+        help="Opt into real upstream calls (default: recorded, offline, $0).",
+    )
+    p.add_argument(
+        "--base-url",
+        default=None,
+        help="Pin the live target host (needed when a local spec's server can't be trusted).",
+    )
+    args = p.parse_args(argv)
+    if _reject_unsafe(args.spec, "verify-docs"):
+        return 2
+    try:
+        spec = load_spec(args.spec)
+    except (UnsafeUrlError, OSError, ValueError) as exc:
+        print(f"  ✗ could not read spec at {args.spec}: {exc}", file=sys.stderr)
+        return 2
+
+    mode: CallMode = "live" if args.live else "recorded"
+    client = AgentApiClient(spec, base_url=args.base_url, session=stub_session())
+    report = verify_mod.verify_docs(client, mode=mode)
+    print(json.dumps(report, indent=2))
+    return 0
+
+
 def _cmd_test(argv: list[str]) -> int:
     p = argparse.ArgumentParser(
         prog="gecko test",
@@ -1433,6 +1478,8 @@ def main(argv: list[str] | None = None) -> int:
         return _cmd_inspect(rest)
     if cmd == "report":
         return _cmd_report(rest)
+    if cmd == "verify-docs":
+        return _cmd_verify_docs(rest)
     if cmd == "from-docs":
         return _cmd_from_docs(rest)
     if cmd == "scan-image":

gecko/graph.py

@@ -32,7 +32,7 @@
 from collections import defaultdict
 from collections.abc import Mapping
 from dataclasses import dataclass, field
-from typing import Literal
+from typing import Any, Literal
 
 from .ingest import Operation, Param
 from .sanitize import key_is_dangerous
@@ -138,6 +138,52 @@ class Plan:
     explain: tuple[ExplainEntry, ...]
 
 
+# --- provenance classification: the CLAIMED entry point (single source of truth) ---
+# gecko.docs_reader stamps this on ``info`` when it recovers a draft OpenAPI from a human
+# page; the per-op keys are its confidence/review annotations. Either marks an op as a
+# claim an untrusted docs source made — provenance CLAIMED — until reality answers.
+_DOCS_DRAFT_MARKER = "gecko.docs_reader"
+_DRAFT_OP_KEYS = ("x-draft-confidence", "x-review")
+
+
+def op_provenance(spec: Mapping[str, Any], operation_id: str) -> Provenance:
+    """Classify one op's provenance from its spec markers: CLAIMED if it was recovered
+    from an untrusted docs source (a from-docs draft), else EXTRACTED. Pure/surface-only
+    — reads only the draft markers, never a schema value.
+
+    This lives here (not in the verify driver) because CLAIMED is the canonical top of
+    the trust ladder (§13.2) and its value has a single home. It is a SEPARATE axis from
+    the ``VerifyVerdict`` a replay later attaches: a CLAIMED op keeps its provenance and
+    gains a verdict; an EXTRACTED op keeps EXTRACTED and gains a verdict just the same.
+    """
+    if not isinstance(spec, Mapping):
+        return "EXTRACTED"
+    info = spec.get("info")
+    if isinstance(info, Mapping) and info.get("x-generated-by") == _DOCS_DRAFT_MARKER:
+        return "CLAIMED"
+    item = _operation_item(spec, operation_id)
+    if item is not None and any(k in item for k in _DRAFT_OP_KEYS):
+        return "CLAIMED"
+    return "EXTRACTED"
+
+
+def _operation_item(
+    spec: Mapping[str, Any], operation_id: str
+) -> Mapping[str, Any] | None:
+    """The raw path-item operation object for ``operation_id``, or None — read ONLY for
+    its draft markers (provenance), never a schema value."""
+    paths = spec.get("paths")
+    if not isinstance(paths, Mapping):
+        return None
+    for path_item in paths.values():
+        if not isinstance(path_item, Mapping):
+            continue
+        for op in path_item.values():
+            if isinstance(op, Mapping) and op.get("operationId") == operation_id:
+                return op
+    return None
+
+
 # --- name / entity helpers (ported from the v3 probe) ---------------------------
 def _norm(s: str) -> str:
     return s.replace("_", "").replace("-", "").lower()

gecko/verify.py

@@ -0,0 +1,140 @@
+"""verify-docs — check a surface's operations against reality, attach verdicts (VAS-2).
+
+VAS-1 shipped the vocabulary (the claimed-until-checked provenance tier,
+``graph.VerifyVerdict``, ``validator.verdict_from_replay``). This module is the driver that
+walks every tool on a surface, replays it through the ALREADY-SHIPPED call path
+(``client.call`` → ``caller`` / ``validator`` — carrying the auth-host pin, recorded-mode
+scrub, and SSRF guard, so no new network path is introduced), lifts the outcome into a
+``VerifyVerdict``, and attaches it to the operation's graph node (``Node.verify``) as a
+SEPARATE axis from provenance (``graph.op_provenance`` is the single home of that axis).
+
+Two modes, one code path (invariant #3):
+
+* ``recorded`` (default): the response is synthesized from the schema, so no status ever
+  comes off the wire. Every op is honestly **UNVERIFIED** — the offline, $0,
+  falsifiable baseline. A recorded 200 is NEVER overclaimed as VERIFIED (the honesty
+  flag lives in ``verdict_from_replay``: only ``source == "observed"`` can VERIFY/REFUTE).
+* ``live``: the real upstream is called. A 2xx/3xx VERIFIES the op; a 404 (or any 4xx/5xx)
+  REFUTES it — the fabricated-endpoint case (a docs source asserting an endpoint the API
+  does not serve). If the surface degrades live→recorded (quarantine / no injectable auth),
+  the actual mode is synthetic and the op stays UNVERIFIED (honest "no access").
+
+Control plane only (invariant #1): the returned report carries op ids + verdict STATUS +
+BASIS strings + a provenance category + counts. It never carries a response body, an arg
+value, a filled URL, or a secret — none of those ever enter this module (the verdict is
+lifted from a payload-free ``CallOutcome``, and the synthesized response ``data`` from a
+recorded call is read for its STATUS only, never copied out).
+"""
+
+from __future__ import annotations
+
+import time
+from typing import TYPE_CHECKING, Any
+
+from . import corpus
+from .caller import CallError
+from .graph import VerifyStatus, op_provenance
+from .modes import CallMode
+from .validator import example_args, verdict_from_replay
+
+if TYPE_CHECKING:
+    from .client import AgentApiClient
+
+# op_provenance (the claimed-until-checked classifier) lives in graph.py — its home so the
+# provenance ladder has a single source of truth; re-exported here as the verify surface.
+__all__ = ["op_provenance", "verify_docs"]
+
+_STATUS_TO_SUMMARY: dict[VerifyStatus, str] = {
+    "VERIFIED": "verified",
+    "REFUTED": "refuted",
+    "UNVERIFIED": "unverified",
+}
+
+
+def _replay_outcome(
+    client: AgentApiClient, tool: dict[str, Any], mode: CallMode
+) -> corpus.CallOutcome:
+    """Replay one tool through the shipped call path and lift the result into a
+    control-plane-safe ``CallOutcome`` (metadata only — status/shape, never a body).
+
+    Reuses ``client.call`` so the auth-host pin, recorded-mode scrub, and SSRF guard all
+    still apply (no new network path). The ACTUAL mode returned by ``call`` — not the
+    requested one — feeds the outcome's source: a live call that degraded to recorded
+    (quarantine / no injectable auth) is honestly synthetic, so it can never VERIFY.
+    """
+    tool_name = tool["name"]
+    op = client._op_by_name[tool_name]
+    args = example_args(tool)
+    invoke = tool.get("_invoke")
+    if not isinstance(invoke, dict):
+        invoke = {"method": op.method, "path": op.path}
+
+    status: int | None = None
+    exc: CallError | None = None
+    actual_mode: str = mode
+    try:
+        result = client.call(tool_name, args, mode=mode)
+        status = result.get("status")
+        # honest source: a live call that degraded shows its real (recorded) mode here.
+        actual_mode = result.get("mode", mode)
+    except CallError as err:  # pre-flight failure (missing input / auth-host refusal)
+        exc = err
+
+    return corpus.outcome_from(
+        operation_id=op.operation_id,
+        tool_invoke=invoke,
+        args=args,
+        status=status,
+        error_class=corpus.error_class_for(status, exc),
+        latency_ms=None,
+        mode=actual_mode,
+        auth_injected=False,
+        ts=int(time.time() * 1000),
+        surface_id=client.surface_id,
+        surface_rev=client.surface_rev,
+    )
+
+
+def verify_docs(
+    client: AgentApiClient, *, mode: CallMode = "recorded"
+) -> dict[str, Any]:
+    """Verify every operation on the surface against reality and attach the verdicts.
+
+    For each usable tool: replay it (``_replay_outcome``), lift the outcome with VAS-1's
+    ``verdict_from_replay``, attach the ``VerifyVerdict`` to the op's graph node
+    (``Node.verify`` — a separate axis; provenance is left untouched), and record a
+    control-plane-safe entry ``{status, basis, provenance}`` in the report.
+
+    Returns ``{"mode", "report": {op_id: {...}}, "summary": {verified, refuted,
+    unverified}}``. Recorded mode (default) → every op UNVERIFIED (no wire evidence);
+    live mode → real VERIFIED / REFUTED. Never returns a payload, value, or secret.
+    """
+    graph = client.surface_graph
+    report: dict[str, dict[str, Any]] = {}
+    summary = {"verified": 0, "refuted": 0, "unverified": 0}
+
+    for tool in client.list_tools():
+        op = client._op_by_name.get(tool["name"])
+        if op is None:  # a synthetic tool with no backing operation — nothing to verify
+            continue
+        op_id = op.operation_id
+        outcome = _replay_outcome(client, tool, mode)
+        verdict = verdict_from_replay(outcome)
+
+        # Attach the verdict to the op node. Node is frozen because its identity (the
+        # content hash) is derived from the SHAPE; ``verify`` is deliberately excluded
+        # from that hash (a drift-dependent replay result), so setting it post-hoc via
+        # object.__setattr__ is the intended out-of-hash mutation, not a contract break.
+        node = graph._by_id.get(graph.opnode(op_id))
+        if node is not None:
+            object.__setattr__(node, "verify", verdict)
+
+        report[op_id] = {
+            "status": verdict.status,
+            "basis": list(verdict.basis),
+            # provenance is a SEPARATE axis — never overwritten by the verdict.
+            "provenance": op_provenance(client.spec, op_id),
+        }
+        summary[_STATUS_TO_SUMMARY[verdict.status]] += 1
+
+    return {"mode": mode, "report": report, "summary": summary}