Pup - the AI's own browser

#!/usr/bin/env python3
"""not_intercepted.py — the SDK's NOT-INTERCEPTED audit (adom/pup-bridge#61, ab#86).

ab is supposed to FORWARD a bridge's verbs and relay the response verbatim. When ab core carries its
own implementation of one of your verbs, nothing errors: the caller gets a plausible, well-formed
answer that is not yours, and the blame lands on your repo. The tell is the KEY SET.

For each verb this calls pup's OWN port directly (POST /command) and the same verb through
adom-bridge, then diffs the CONTENT keys. ab lifts pup's {success, output:"<json>"} envelope to top
level by design, so both sides are unwrapped before comparing — otherwise every verb looks wrong and
real interception hides in the noise.

    tools/not_intercepted.py [TARGET]        # default AdomLapper

Run after ANY ab upgrade. Read-only verbs only; it must never mutate the user's desktop.

Two hard-won rules live in this file:
  * a probe that could not run is NOT a pass — it is reported and makes the run inconclusive;
  * the request body is verified on disk before the direct call, because a stale body file made an
    earlier version compare pup_status against every verb and report six false mismatches.
"""
import json
import subprocess
import sys

TARGET = sys.argv[1] if len(sys.argv) > 1 else "AdomLapper"
CLI = ["adom-bridge", "--target", TARGET]
BODY_WIN = r"C:\Users\john\AppData\Local\Temp\pup-audit-body.json"


def cli(verb, args, timeout=90):
    try:
        p = subprocess.run(CLI + [verb, json.dumps(args)], capture_output=True, text=True, timeout=timeout)
        raw = p.stdout or ""
        i = raw.find("{")
        return json.loads(raw[i:]) if i >= 0 else {}
    except Exception:
        return {}


def content_keys(payload):
    """Keys that describe the RESPONSE CONTENT, with the output envelope unwrapped and ab's
    transport/telemetry additions removed."""
    ENVELOPE = {
        "durationMs", "status", "success", "error", "aiThread", "window", "sessionId", "ok",
        "timeoutSeconds", "_hint", "_next", "related", "_related", "_cacheHint", "narrative",
        "_reportIssues", "_timeoutHint", "output",
        # ab adds statusVerb from the bridge manifest on its way out — a documented transport
        # addition, present on verbs it demonstrably forwards, so it is not an interception tell.
        "statusVerb",
    }
    keys = set(payload.keys())
    out = payload.get("output")
    if isinstance(out, str):
        try:
            keys |= set(json.loads(out).keys())
        except Exception:
            pass
    return keys - ENVELOPE


def direct_call(port, verb, args):
    """Call pup's own port. The body goes to a file (quoting a JSON body through bash -> CLI JSON ->
    cmd defeated three attempts), and the file is READ BACK to prove it holds THIS verb."""
    body = json.dumps({"command": verb, "args": args})
    cli("desktop_write_file", {"path": BODY_WIN, "content": body,
                               "reason": f"NOT-INTERCEPTED audit: stage the request body for {verb}"})
    check = cli("shell_execute", {"command": r"type %TEMP%\pup-audit-body.json",
                                  "reason": "NOT-INTERCEPTED audit: verify the staged body"})
    staged = (check.get("stdout") or "").strip()
    if verb not in staged:
        return None, f"body file does not contain {verb} (staged: {staged[:60]!r})"
    r = cli("shell_execute", {
        "command": (r'curl.exe -s -X POST -H "Content-Type: application/json" '
                    rf"--data-binary @%TEMP%\pup-audit-body.json http://127.0.0.1:{port}/command"),
        "reason": f"NOT-INTERCEPTED audit: call pup's own port directly for {verb}"})
    out = (r.get("stdout") or "").strip()
    if not out:
        return None, "empty response from pup's port"
    try:
        return json.loads(out), None
    except Exception as e:
        return None, f"unparseable: {e}"


def main():
    bl = cli("bridge_list", {})
    port = None
    for b in bl.get("bridges") or []:
        if isinstance(b, dict) and b.get("name") == "puppeteer":
            port = (b.get("spawn") or {}).get("runtimePort") or b.get("port")
    if not port:
        print(f"could not resolve pup's port on {TARGET}")
        return 1
    print(f"== NOT-INTERCEPTED audit on {TARGET} (pup port {port}) ==")

    sid = "ni-audit-probe"
    cli("pup_open_window", {"sessionId": sid, "url": "about:blank", "foreground": False}, timeout=150)

    verbs = [
        ("pup_status", {}), ("pup_readiness", {}), ("pup_list_windows", {}), ("pup_describe", {}),
        ("pup_window_info", {"sessionId": sid}), ("pup_list_tabs", {"sessionId": sid}),
        ("pup_lower_os_window", {"sessionId": sid}),
        # no foregroundReason on purpose: both sides should REFUSE identically, and a refusal that
        # differs is as telling as a success that differs
        ("pup_raise_os_window", {"sessionId": sid}),
        ("pup_alert_window", {"sessionId": sid, "stop": True}),
        ("pup_progress", {"sessionId": sid, "mode": "off"}),
    ]
    bad, skipped, findings = 0, 0, []
    for verb, args in verbs:
        d, err = direct_call(port, verb, args)
        if d is None:
            print(f"  ?? {verb} — DIRECT CALL FAILED ({err}); probe did not run, this is NOT a pass")
            skipped += 1
            continue
        routed = cli(verb, args)
        dk, rk = content_keys(d), content_keys(routed)
        only_ab, only_pup = rk - dk, dk - rk
        if only_ab or only_pup:
            bad += 1
            findings.append((verb, sorted(only_ab), sorted(only_pup)))
            print(f"  MISMATCH {verb}")
            if only_ab:
                print(f"      only via ab  (ab may be answering): {sorted(only_ab)}")
            if only_pup:
                print(f"      only direct  (ab dropped these):   {sorted(only_pup)}")
        else:
            print(f"  ok {verb} — {len(dk)} content keys, identical both ways")

    cli("pup_close_window", {"sessionId": sid})
    checked = len(verbs) - skipped
    print()
    if skipped:
        print(f"== INCONCLUSIVE: {checked} of {len(verbs)} compared, {skipped} probe(s) failed to run ==")
        return bad + skipped
    if bad:
        print(f"== {bad} of {checked} verbs differ — report these on the ab issue ==")
    else:
        print(f"== CLEAN: ab forwards all {checked} audited verbs verbatim ==")
    return bad


if __name__ == "__main__":
    sys.exit(main())