openswarm/scripts/browser_canary.py

"""Drift canary for the browser write path.

WHY THIS EXISTS: on 2026-07-21 X renamed its compose control and our sends silently went to 0/2
WITH false success claims. No code of ours changed. A passing test suite is a snapshot; sites move
underneath it, so the only way a coverage claim stays true is to re-prove it against the live site
on a schedule and shout the day it breaks.

WHAT IT DOES: per site, one full round trip on the user's own account, then cleans up after itself:
    post a unique marker -> confirm the receipt -> delete it -> confirm it is gone
A site "passes" only if the post was receipt-verified AND the cleanup verified gone. Anything else
is drift, reported with the stage it died at.

SAFETY:
  - Never runs by itself. No cron, no import side effects; a human or a CI job invokes it.
  - --dry (default) posts NOTHING: it exercises discovery only, so it is safe anywhere.
  - --live is required to actually post, and every post is deleted in the same run.
  - Markers are random and carry no removal words ("delete"/"remove" in a payload trips the
    removal classifier and makes the send path stand down, which cost us a confusing hour once).
  - Exits 1 on drift so a scheduler can alert on it.

USAGE
    python scripts/browser_canary.py                     # discovery only, posts nothing
    python scripts/browser_canary.py --live              # real round trip, self-cleaning
    python scripts/browser_canary.py --live --sites x,reddit
    (backend must be running; set OSW_CANARY_BASE if it is not on :8326)
"""

import argparse
import json
import os
import re
import secrets
import sys
import time
import urllib.request
from typing import Dict, List, Optional

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
BASE = os.environ.get("OSW_CANARY_BASE", "http://127.0.0.1:8326") + "/api/agents"
LOG = os.environ.get("OSW_CANARY_LOG", "/tmp/osw_backend_mr.log")
MODEL = os.environ.get("OSW_CANARY_MODEL", "opus-4-8")

# Per site: how to post it, how to delete it. Kept deliberately small: these are the surfaces we
# CLAIM to support, so the canary's job is to keep that claim honest, not to explore new ones.
SITES: Dict[str, Dict[str, str]] = {
    "x": {
        "probe": 'Go to x.com. Do NOT type or post anything. Is the tweet compose box present on the page? Answer with exactly one word: YES or NO.',
        "post": 'Go to x.com and post this tweet, exactly: "{m}"',
        "delete": 'Go to x.com/{handle} and delete the post that says "{m}"',
        "verify": ('Go to x.com/{handle}. Do NOT post, delete or change anything. Is there a post '
                   'containing "{m}"? Answer with exactly one word: GONE if absent, PRESENT if there.'),
        "handle_env": "OSW_CANARY_X_HANDLE",
    },
    "linkedin": {
        "probe": 'Go to linkedin.com. Do NOT type or post anything. Is the "Start a post" compose control present on the feed? Answer with exactly one word: YES or NO.',
        "post": 'Go to linkedin.com and create a post with exactly this text: "{m}"',
        "delete": 'Go to my LinkedIn activity and delete the post that says "{m}"',
        "verify": ('Go to my LinkedIn activity. Do NOT post, delete or change anything. Is there a post '
                   'containing "{m}"? Answer with exactly one word: GONE if absent, PRESENT if there.'),
    },
    "reddit": {
        "probe": 'Go to reddit.com/r/test/submit. Do NOT type or submit anything. Is the post title/body compose form present? Answer with exactly one word: YES or NO.',
        "post": 'Go to reddit.com and create a text post in r/test titled "{m}" with body "canary check". Submit it.',
        "delete": 'Go to reddit.com and delete my post titled "{m}"',
        "verify": ('Go to my reddit profile. Do NOT post, delete or change anything. Is there a post '
                   'titled "{m}"? Answer with exactly one word: GONE if absent, PRESENT if there.'),
    },
}


def req(method: str, url: str, body: Optional[dict] = None) -> dict:
    tok = open(os.path.join(ROOT, "backend/data/auth.token")).read().strip()
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(url, data=data, method=method,
                               headers={"Content-Type": "application/json",
                                        "Authorization": "Bearer " + tok})
    with urllib.request.urlopen(r, timeout=260) as resp:
        return json.loads(resp.read().decode() or "{}")


def log_lines() -> int:
    try:
        return sum(1 for _ in open(LOG, errors="ignore"))
    except OSError:
        return 0


def run_task(prompt: str, name: str, budget: int = 180) -> Dict[str, object]:
    """Dispatch one browser task; return {said, status, wall, log} for the slice it produced."""
    mark = log_lines()
    dash = req("GET", BASE.replace("/agents", "") + "/dashboards/list")
    dashboards = dash if isinstance(dash, list) else dash.get("dashboards", [])
    if not dashboards:
        return {"said": "", "status": "error", "wall": 0.0, "log": "", "err": "no dashboard"}
    sid = req("POST", f"{BASE}/launch", {"mode": "agent", "model": MODEL, "provider": "anthropic",
                                         "dashboard_id": dashboards[0]["id"], "name": name})["session"]["id"]
    t0 = time.time()
    try:
        req("POST", f"{BASE}/sessions/{sid}/message", {"prompt": prompt, "mode": "agent", "model": MODEL})
    except Exception:
        pass
    status = ""
    said = ""
    while time.time() - t0 < budget:
        try:
            s = req("GET", f"{BASE}/sessions/{sid}")
            status = str(s.get("status") or "")
            if status in ("completed", "error", "stopped"):
                msgs = [m for m in s.get("messages", []) if m.get("role") == "assistant"]
                if msgs:
                    c = msgs[-1].get("content")
                    said = c if isinstance(c, str) else str(c)
                break
        except Exception:
            pass
        time.sleep(1.5)
    time.sleep(1.0)
    try:
        slice_ = "".join(open(LOG, errors="ignore").readlines()[mark:])
    except OSError:
        slice_ = ""
    return {"said": said, "status": status, "wall": round(time.time() - t0, 1), "log": slice_}


def check_site(site: str, cfg: Dict[str, str], live: bool) -> Dict[str, object]:
    marker = "canary" + secrets.token_hex(4)          # no removal words, unique per run
    res: Dict[str, object] = {"site": site, "marker": marker, "live": live}

    if not live:
        # Discovery-only, safe BY CONSTRUCTION rather than by configuration: we ask the agent to
        # REPORT whether the compose surface is reachable, never to write. An earlier draft set
        # OSW_SENDSCRIPT_DRYRUN in this process, which does nothing to the backend, so a "dry" run
        # against a normal backend would have posted for real. Never trust a flag you don't own.
        r = run_task(cfg["probe"].format(handle=os.environ.get(cfg.get("handle_env", ""), "")),
                     f"canary-probe-{site}")
        said = str(r["said"])
        found = bool(re.search(r"\bYES\b", said)) and not re.search(r"\bNO\b", said)
        if "sent_receipt=True" in str(r["log"]):        # must never happen on this path
            res.update(stage="SAFETY", ok=False,
                       detail="ABORT: a discovery probe performed a real send; check the task text")
            return res
        res.update(stage="discovery", ok=found, wall=r["wall"],
                   detail=("compose surface reachable" if found else f"NOT reachable: {said[:110]}"))
        return res

    # 1. POST, and require the two-sided receipt (composer cleared), not the model's word for it.
    handle = os.environ.get(cfg.get("handle_env", ""), "") if cfg.get("handle_env") else ""
    r = run_task(cfg["post"].format(m=marker, handle=handle), f"canary-post-{site}")
    log = str(r["log"])
    delivered = "done sent_receipt=True" in log or "DELIVERY CONFIRMED" in log
    res["post_wall"] = r["wall"]
    if not delivered:
        res.update(stage="post", ok=False,
                   detail=f"no receipt (status={r['status']}): {str(r['said'])[:120]}")
        return res

    # 2. DELETE, and require the in-page verify-gone, so cleanup can't be claimed falsely.
    d = run_task(cfg["delete"].format(m=marker, handle=handle), f"canary-clean-{site}")
    dlog = str(d["log"])
    removed = "removed=True" in dlog
    res["delete_wall"] = d["wall"]
    if not removed and cfg.get("verify"):
        # `removed=True` only exists on the BrowserDeleteItem dispatch path. A model-driven delete is
        # every bit as real, and grepping for the mechanism called DRIFT on a delete that had plainly
        # worked: X said "Your post was deleted" and an independent profile audit found the marker
        # gone, twice. A canary that cries wolf gets ignored, which is worse than no canary, so the
        # authority here is the OUTCOME (is it still on the page?) and never the implementation detail.
        v = run_task(cfg["verify"].format(m=marker, handle=handle), f"canary-verify-{site}")
        vsaid = str(v["said"])
        removed = bool(re.search(r"\bGONE\b", vsaid, re.I)) and not re.search(r"\bPRESENT\b", vsaid, re.I)
        res["verify_wall"] = v["wall"]
        res["verified_by"] = "re-read" if removed else "re-read (still present)"
    if not removed:
        res.update(stage="cleanup", ok=False,
                   detail=f"POSTED BUT NOT CLEANED, marker {marker} may be live: {str(d['said'])[:100]}")
        return res
    res.update(stage="done", ok=True, detail="posted, receipt-verified, deleted, verified gone")
    return res


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--live", action="store_true",
                    help="actually post (and delete) on the real accounts; default is discovery-only")
    ap.add_argument("--sites", default="", help="comma list (default: all)")
    args = ap.parse_args()

    want: List[str] = [s.strip() for s in args.sites.split(",") if s.strip()] or list(SITES)
    unknown = [s for s in want if s not in SITES]
    if unknown:
        print(f"unknown site(s): {', '.join(unknown)}; known: {', '.join(SITES)}")
        return 2

    print(f"browser drift canary  mode={'LIVE (posts+deletes)' if args.live else 'discovery-only'}  "
          f"sites={','.join(want)}  model={MODEL}")
    rows = []
    for s in want:
        r = check_site(s, SITES[s], args.live)
        rows.append(r)
        flag = "PASS" if r.get("ok") else "DRIFT"
        print(f"  [{flag}] {s:10} stage={r.get('stage','?'):10} {r.get('detail','')}", flush=True)

    bad = [r for r in rows if not r.get("ok")]
    stranded = [r for r in bad if r.get("stage") == "cleanup"]
    print(f"\n{len(rows) - len(bad)}/{len(rows)} sites healthy")
    if stranded:
        print("!! MANUAL CLEANUP NEEDED: " + ", ".join(f"{r['site']}:{r['marker']}" for r in stranded))
    if bad:
        print("DRIFT DETECTED on: " + ", ".join(str(r["site"]) for r in bad))
    return 1 if bad else 0


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