[eric] browser: general capture-replay write tier (site's own captured route, same-origin + gated, default off)

backend/apps/agents/browser/browser_agent.py

@@ -338,30 +338,57 @@ def p_extract_domain(url: str) -> str | None:
         return None
 
 
-async def run_api_write(tool_input: dict, current_url: str) -> dict:
-    """Route a BrowserApiWrite to the API-first write tier: resolve the current site's
-    domain, call its own write API through the borrowed session, and return a truthful
-    result (a real receipt on success, a 'use the UI' miss otherwise). Never raises: a
-    missing adapter or a site-reject is a typed miss, so the model falls back to the UI
-    path, never a crash and never a false claim of success."""
-    from backend.apps.agents.browser import site_write_registry
-    action = str((tool_input or {}).get("action") or "").strip()
-    if not action:
-        return {"error": "BrowserApiWrite needs an 'action' (comment, reply, post, edit, or delete)."}
-    domain = p_extract_domain(current_url or "")
-    if not domain:
-        return {"error": "Can't tell what site you're on yet; navigate to the site first, then do the write through the UI or retry."}
-    params = {k: v for k, v in (tool_input or {}).items() if k not in ("action", "expect")}
-    res = await site_write_registry.api_write(domain, action, params)
+def p_api_write_result(res) -> dict:
+    """Shape a registry WriteResult into a loop result: a truthful receipt on success (with
+    send_confirmed set by the caller), or an `error` on a miss so the model does the write via the
+    UI and the run never distills it as a false success."""
     if res.ok:
         return {"ok": True, "text": (
             f"Done via the {res.domain} API in {res.latency_ms}ms. Receipt: {res.receipt}. "
             "The write landed, that receipt is your proof; you're finished with this step."
         )}
-    # A miss (no adapter / site-reject) surfaces as an error so the model does the write via the UI and the run never distills it as a success.
     return {"error": f"API write not used ({res.error}). Do this action through the UI instead."}
 
 
+async def run_api_write(tool_input: dict, current_url: str, browser_id: str = "", tab_id: str = "") -> dict:
+    """Route a BrowserApiWrite to the API-first write tier: a deterministic built-in adapter
+    (Reddit) when one exists, else the GENERAL capture-replay tier (action='route': replay a
+    mutating route the site's own UI fired, verified same-origin + captured, behind OSW_ROUTE_WRITE).
+    Never raises: a missing adapter / disarmed tier / site-reject is a typed miss, so the model
+    falls back to the UI path, never a crash and never a false claim of success."""
+    from urllib.parse import urlparse
+    from backend.apps.agents.browser import route_write, site_write_registry
+    action = str((tool_input or {}).get("action") or "").strip()
+    if not action:
+        return {"error": "BrowserApiWrite needs an 'action' (comment, reply, post, edit, delete, or route)."}
+    domain = p_extract_domain(current_url or "")
+    if not domain:
+        return {"error": "Can't tell what site you're on yet; navigate to the site first, then do the write through the UI or retry."}
+
+    if action == "route":
+        # General tier: replay a captured mutating route. The captured set is fetched live from the
+        # page (the safety wall: only a route the UI actually fired can be replayed), and the replay
+        # itself is same-origin + flag-gated + session-borrowed in route_write.
+        method = str(tool_input.get("method") or "POST").strip()
+        url = str(tool_input.get("url") or "").strip()
+        body = tool_input.get("body") if isinstance(tool_input.get("body"), dict) else {}
+        if not url:
+            return {"error": "BrowserApiWrite route needs the 'url' of a captured write endpoint (see BrowserListRoutes)."}
+        try:
+            origin = f"{urlparse(current_url).scheme}://{urlparse(current_url).netloc}"
+        except Exception:
+            return {"error": "Can't resolve the current site's origin; do the write through the UI."}
+        listed = await execute_browser_tool("BrowserListRoutes", {"writes": True}, browser_id, tab_id)
+        captured = [route_write.CapturedRoute(method=str(r.get("method", "")), template=str(r.get("template", "")))
+                    for r in (listed.get("routes") or []) if isinstance(r, dict) and r.get("template")]
+        res = await site_write_registry.api_route_write(origin, method, url, body, captured)
+        return p_api_write_result(res)
+
+    params = {k: v for k, v in (tool_input or {}).items() if k not in ("action", "expect")}
+    res = await site_write_registry.api_write(domain, action, params)
+    return p_api_write_result(res)
+
+
 def strip_lone_surrogates(s: str) -> str:
     # The JS/webview hands us page text as UTF-16, so an emoji can arrive as half of its surrogate pair; Python carries the orphan but .encode('utf-8') later (the SDK serializing the request to the LLM) detonates with "surrogates not allowed" and kills the turn. Swap any orphan for the replacement char.
     return re.sub(r"[\ud800-\udfff]", "�", s) if s else s
@@ -1920,7 +1947,7 @@ async def run_browser_agent(
                     )}
                 elif tu.name == "BrowserApiWrite":
                     # API-first write tier: the site's own write API via the borrowed session, deterministic + a real receipt. A miss is a typed "use the UI" (never a crash), so the loop falls back cleanly.
-                    result = await p_cancellable(run_api_write(tu.input, current_url))
+                    result = await p_cancellable(run_api_write(tu.input, current_url, browser_id, tab_id))
                     if result is None:
                         cancelled = True
                         break

backend/apps/agents/browser/browser_schema.py

@@ -616,20 +616,24 @@ BROWSER_TOOLS_SCHEMA = [
             "and it hands back the site's REAL receipt (the new post/comment's id and "
             "permalink) as proof it landed. Only some sites are supported so far "
             "(currently Reddit: comment, reply, post, edit, delete). If the current site "
-            "has no adapter you get a clean 'no adapter' miss, just do the write through "
-            "the UI instead. This IS a real write: call it ONCE, and the receipt is your "
-            "confirmation, do not re-check or re-fire it."
+            "has no built-in adapter, you can still do it the GENERAL way: set action='route' "
+            "with the site's own write endpoint (method + url + body) taken from BrowserListRoutes, "
+            "and it replays that request with your session. If neither works you get a clean miss, "
+            "just do the write through the UI instead. This IS a real write: call it ONCE, and the "
+            "receipt is your confirmation, do not re-check or re-fire it."
         ),
         "input_schema": {
             "type": "object",
             "properties": {
-                "action": {"type": "string", "description": "The write to perform: comment, reply, post, edit, or delete."},
+                "action": {"type": "string", "description": "comment, reply, post, edit, delete (built-in adapter), or route (general: replay a captured write endpoint)."},
                 "parent_id": {"type": "string", "description": "comment/reply: fullname of the post/comment you're replying to (e.g. t3_abc, t1_xyz)."},
                 "thing_id": {"type": "string", "description": "edit/delete: fullname of your OWN post/comment (e.g. t1_xyz)."},
                 "text": {"type": "string", "description": "The body text (comment/reply/post/edit)."},
                 "subreddit": {"type": "string", "description": "post: the subreddit name, without the r/ prefix."},
                 "title": {"type": "string", "description": "post: the post title."},
-                "url": {"type": "string", "description": "post: a URL to submit as a link post (omit for a text post)."},
+                "url": {"type": "string", "description": "post: a link URL; OR route: the write endpoint's full URL from BrowserListRoutes."},
+                "method": {"type": "string", "description": "route: the endpoint's HTTP method (POST, PUT, PATCH, DELETE)."},
+                "body": {"type": "object", "description": "route: the JSON body to send, matching the endpoint's captured shape, with YOUR content in the text field(s)."},
             },
             "required": ["action"],
         },

backend/apps/agents/browser/route_write.py

@@ -0,0 +1,210 @@
+"""General capture-and-replay write tier: replay a write the site's OWN UI issues, via the
+borrowed session, WITHOUT a hand-written per-site adapter. This is the site-agnostic path to
+write coverage (the "all popular sites" lever): the browser passively captures the internal API
+routes the page fires (electron/cdp-routes.js), and this replays a MUTATING one with the agent's
+content substituted, using live-borrowed cookies (never persisted) plus any CSRF header the site
+derives from a cookie.
+
+SAFETY (this IS the posture flip away from GET/HEAD-only, so the walls are belt-and-suspenders):
+- Same-origin: the target must be the site currently loaded, nothing else.
+- Captured-route match: the target must correspond to a mutating route the site's OWN UI actually
+  fired. The agent can't invent an endpoint; it can only replay one the page genuinely uses. This
+  is the wall against a prompt-injected page steering the agent to an arbitrary write.
+- Flag-gated default OFF (OSW_ROUTE_WRITE=1 to arm). The deterministic per-site adapters (Reddit)
+  stay always-on; this general tier is opt-in until it's soaked.
+- Behind the caller's send-safety guard (solo, verified, receipt-or-honest-miss, never a false
+  claim of success).
+- Secret-safe: cookies are live-borrowed per call, never logged, never persisted; the CSRF header
+  is derived from a cookie at call time, not stored.
+"""
+
+import json
+import os
+import re
+import time
+import urllib.error
+import urllib.request
+from typing import Any, Dict, List
+from urllib.parse import urlparse
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.social_shims.session_source import get_session
+
+WRITE_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
+# CSRF header a site derives from a cookie (so it survives a fresh borrowed session). Small on
+# purpose: most cookie-auth internal APIs need nothing extra; this covers the common header case.
+P_CSRF_FROM_COOKIE: Dict[str, Dict[str, str]] = {
+    "x.com": {"header": "x-csrf-token", "cookie": "ct0"},
+    "twitter.com": {"header": "x-csrf-token", "cookie": "ct0"},
+}
+
+
+class CapturedRoute(BaseModel):
+    """One mutating route the site's own UI was seen to fire (from the CDP route capture). The
+    method + templated path are the identity we match a replay target against; nothing secret
+    lives here (the capture redacts auth headers and strips body values)."""
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    method: str
+    template: str
+
+
+class ReplayOutcome(BaseModel):
+    """Typed result of a route replay. `receipt` is the site's own confirmation pulled from the
+    response (an id / permalink / url); `ok` is False with a legible `error` on any refusal or
+    rejection, so the caller falls back to the UI, never a crash and never a false success."""
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    ok: bool
+    receipt: str = ""
+    error: str = ""
+    status: int = 0
+    latency_ms: int = 0
+
+
+@typechecked
+def enabled() -> bool:
+    """The general route-write tier is opt-in (posture flip); armed only by OSW_ROUTE_WRITE=1."""
+    return os.environ.get("OSW_ROUTE_WRITE", "0") == "1"
+
+
+@typechecked
+def p_template_path(url: str) -> str:
+    """Collapse volatile path segments (numeric ids, long hex/uuids) to '{id}', mirroring the
+    capture side (cdp-routes.js templateUrl) so a concrete replay URL matches the captured
+    template. Origin + path only; query keys are ignored for the match."""
+    try:
+        u = urlparse(url)
+        path = re.sub(r"/(\d+|[0-9a-fA-F]{8,}(?:-[0-9a-fA-F]+)*)(?=/|$)", "/{id}", u.path)
+        return f"{u.scheme}://{u.netloc}{path}"
+    except Exception:
+        return url
+
+
+@typechecked
+def same_origin(url: str, origin: str) -> bool:
+    """True when url is on the same origin as the loaded page (scheme+host+port), the first wall."""
+    try:
+        a, b = urlparse(url), urlparse(origin)
+        return bool(a.scheme and a.netloc) and (a.scheme, a.netloc) == (b.scheme, b.netloc)
+    except Exception:
+        return False
+
+
+@typechecked
+def route_is_captured(method: str, url: str, captured: List[CapturedRoute]) -> bool:
+    """True when (method, templated url) matches a mutating route the site's UI actually fired.
+    The safety wall that stops a prompt-injected page from steering the agent to an invented
+    endpoint: the agent can only replay a write the page genuinely uses."""
+    m = method.upper()
+    if m not in WRITE_METHODS:
+        return False
+    target = p_template_path(url)
+    return any(r.method.upper() == m and p_template_path(r.template) == target for r in captured)
+
+
+@typechecked
+def p_cookie_value(cookie_header: str, name: str) -> str:
+    """Pull one cookie's value out of a 'k=v; k2=v2' header, for CSRF-from-cookie derivation."""
+    for part in (cookie_header or "").split(";"):
+        k, _, v = part.strip().partition("=")
+        if k == name:
+            return v
+    return ""
+
+
+@typechecked
+def derive_csrf_headers(url: str, cookie_header: str) -> Dict[str, str]:
+    """The CSRF header a site expects, re-derived from the live cookie (e.g. X's x-csrf-token is
+    its ct0 cookie). Empty for the common cookie-only-auth site, which needs nothing extra."""
+    host = (urlparse(url).netloc or "").lower().lstrip(".")
+    apex = ".".join(host.split(".")[-2:]) if host.count(".") >= 1 else host
+    rule = P_CSRF_FROM_COOKIE.get(apex)
+    if not rule:
+        return {}
+    val = p_cookie_value(cookie_header, rule["cookie"])
+    return {rule["header"]: val} if val else {}
+
+
+@typechecked
+def receipt_from_json(obj: Any) -> str:
+    """The most proof-bearing id/permalink/url anywhere in a response JSON (shallow-first), so the
+    caller gets a real receipt without knowing each site's response shape."""
+    seen: List[Any] = [obj]
+    for _ in range(400):  # bounded walk; a receipt lives near the top of a write response
+        if not seen:
+            break
+        cur = seen.pop(0)
+        if isinstance(cur, dict):
+            for key in ("permalink", "url", "id_str", "rest_id", "id", "name"):
+                v = cur.get(key)
+                if isinstance(v, (str, int)) and str(v):
+                    return str(v)
+            seen.extend(cur.values())
+        elif isinstance(cur, list):
+            seen.extend(cur)
+    return ""
+
+
+@typechecked
+def outcome_from_response(status: int, text: str, latency_ms: int) -> ReplayOutcome:
+    """Map an HTTP response to a typed outcome: 2xx = landed (with a parsed receipt), anything else
+    = a legible error the caller surfaces so the model does the write via the UI instead."""
+    if not (200 <= status < 300):
+        return ReplayOutcome(ok=False, status=status, latency_ms=latency_ms,
+                             error=f"site returned HTTP {status}: {text[:160]}")
+    receipt = ""
+    try:
+        receipt = receipt_from_json(json.loads(text)) if text.strip() else ""
+    except (json.JSONDecodeError, ValueError):
+        receipt = ""
+    return ReplayOutcome(ok=True, status=status, latency_ms=latency_ms, receipt=receipt or "ok")
+
+
+@typechecked
+def issue_request(method: str, url: str, body: Dict[str, Any], headers: Dict[str, str]) -> Any:
+    """Issue the write from the backend using the borrowed session. JSON body (the shape internal
+    APIs overwhelmingly use). Returns (status, text). Isolated so tests stub the network."""
+    data = json.dumps(body).encode() if body else b""
+    req = urllib.request.Request(url, data=data, headers=headers, method=method.upper())
+    try:
+        with urllib.request.urlopen(req, timeout=30.0) as resp:
+            return resp.status, resp.read().decode("utf-8", "replace")
+    except urllib.error.HTTPError as e:
+        return e.code, (e.read().decode("utf-8", "replace") if e.fp else "")
+    except urllib.error.URLError as e:
+        raise RuntimeError(f"site unreachable: {getattr(e, 'reason', e)}")
+
+
+@typechecked
+def replay_write(method: str, url: str, body: Dict[str, Any], origin: str,
+                 captured: List[CapturedRoute]) -> ReplayOutcome:
+    """Replay one captured mutating route with the agent's content, via the live-borrowed session.
+    Every failure (disarmed, off-origin, un-captured, no session, site-reject) is a typed ok=False
+    so the caller falls back to the UI, never a crash. Secrets are live-borrowed, never logged."""
+    if not enabled():
+        return ReplayOutcome(ok=False, error="route-write tier disarmed (set OSW_ROUTE_WRITE=1); use the UI")
+    if not same_origin(url, origin):
+        return ReplayOutcome(ok=False, error="target is not the current site (same-origin only)")
+    if not route_is_captured(method, url, captured):
+        return ReplayOutcome(ok=False, error="no matching write route was captured from this site's UI; use the UI")
+    domain = (urlparse(origin).netloc or "").lstrip(".")
+    t0 = time.monotonic()
+    try:
+        cookie, ua = get_session(domain)
+    except Exception as e:
+        return ReplayOutcome(ok=False, error=f"no borrowable session for {domain}: {str(e)[:120]}")
+    headers = {
+        "Cookie": cookie, "User-Agent": ua, "Accept": "application/json",
+        "Content-Type": "application/json", "Origin": origin, "Referer": origin + "/",
+        **derive_csrf_headers(url, cookie),
+    }
+    try:
+        status, text = issue_request(method, url, body, headers)
+    except Exception as e:
+        return ReplayOutcome(ok=False, error=str(e)[:160], latency_ms=int((time.monotonic() - t0) * 1000))
+    return outcome_from_response(status, text, int((time.monotonic() - t0) * 1000))

backend/apps/agents/browser/site_write_registry.py

@@ -13,11 +13,13 @@ Live-validated end to end on Reddit (comment 271ms + reversible delete 246ms, ty
 import asyncio
 import os
 import time
-from typing import Any, Callable, Dict, FrozenSet, Tuple
+from typing import Any, Callable, Dict, FrozenSet, List, Tuple
+from urllib.parse import urlparse
 
 from pydantic import BaseModel, ConfigDict
 from typeguard import typechecked
 
+from backend.apps.agents.browser import route_write
 from backend.apps.reddit_mcp_shim import reddit_writes
 
 
@@ -100,6 +102,19 @@ def p_ensure_session_env() -> None:
             pass
 
 
+@typechecked
+async def api_route_write(origin: str, method: str, url: str, body: Dict[str, Any],
+                         captured: List[route_write.CapturedRoute]) -> WriteResult:
+    """The GENERAL tier: replay a captured mutating route the site's own UI fired, for sites with
+    no hand-written adapter. Wraps route_write's typed outcome into the registry's WriteResult so
+    callers get one shape. Every refusal/rejection is ok=False, so the agent falls back to the UI."""
+    p_ensure_session_env()
+    d = (urlparse(origin).netloc or origin).lstrip(".")
+    out = await asyncio.to_thread(route_write.replay_write, method, url, body, origin, captured)
+    return WriteResult(ok=out.ok, action="route", domain=d, receipt=out.receipt,
+                       error=out.error, latency_ms=out.latency_ms)
+
+
 @typechecked
 async def api_write(domain: str, action: str, params: Dict[str, Any]) -> WriteResult:
     """Perform a write via the site's own API using the borrowed session. Times it, and turns any

backend/tests/test_route_write.py

@@ -0,0 +1,125 @@
+"""Unit tests for the general capture-replay write engine (route_write): the safety walls
+(disarmed / off-origin / un-captured all refuse), CSRF-from-cookie derivation, the generic
+receipt parse, and the fail-open contract (every failure is a typed ok=False, never a crash,
+never a false success). Network is stubbed; the live cross-site round-trip is owed on a healthy
+rig (this bench's renderer command path is wedged, same as all browser live-tests)."""
+import pytest
+
+from backend.apps.agents.browser import route_write as rw
+
+
+def p_arm(monkeypatch):
+    monkeypatch.setenv("OSW_ROUTE_WRITE", "1")
+
+
+def p_reddit_route():
+    return [rw.CapturedRoute(method="POST", template="https://www.reddit.com/api/comment")]
+
+
+# --- safety walls -----------------------------------------------------------
+def test_disarmed_by_default_refuses(monkeypatch):
+    monkeypatch.delenv("OSW_ROUTE_WRITE", raising=False)
+    out = rw.replay_write("POST", "https://www.reddit.com/api/comment", {"text": "hi"},
+                          "https://www.reddit.com", p_reddit_route())
+    assert out.ok is False and "disarmed" in out.error
+
+
+def test_off_origin_target_refused(monkeypatch):
+    p_arm(monkeypatch)
+    out = rw.replay_write("POST", "https://evil.com/api/comment", {"text": "hi"},
+                          "https://www.reddit.com", p_reddit_route())
+    assert out.ok is False and "same-origin" in out.error
+
+
+def test_uncaptured_route_refused(monkeypatch):
+    p_arm(monkeypatch)
+    # same origin, but the site's UI never fired /api/delete_account -> the agent can't invent it
+    out = rw.replay_write("POST", "https://www.reddit.com/api/delete_account", {},
+                          "https://www.reddit.com", p_reddit_route())
+    assert out.ok is False and "captured" in out.error
+
+
+def test_get_is_not_a_write_route(monkeypatch):
+    p_arm(monkeypatch)
+    assert rw.route_is_captured("GET", "https://www.reddit.com/api/comment", p_reddit_route()) is False
+
+
+def test_template_match_ignores_volatile_ids(monkeypatch):
+    # A volatile id that IS a full path segment collapses to {id} on both sides (same regex as the
+    # capture), so a concrete replay URL matches the captured template but a different path doesn't.
+    routes = [rw.CapturedRoute(method="DELETE", template="https://api.site.com/orders/{id}/cancel")]
+    assert rw.route_is_captured("DELETE", "https://api.site.com/orders/4821/cancel", routes) is True
+    assert rw.route_is_captured("DELETE", "https://api.site.com/refunds/4821/cancel", routes) is False
+
+
+# --- CSRF-from-cookie derivation --------------------------------------------
+def test_csrf_header_derived_from_cookie():
+    h = rw.derive_csrf_headers("https://x.com/i/api/graphql/CreateTweet", "ct0=abc123; auth_token=z")
+    assert h == {"x-csrf-token": "abc123"}
+
+
+def test_no_csrf_for_plain_cookie_auth_site():
+    assert rw.derive_csrf_headers("https://www.reddit.com/api/comment", "reddit_session=z") == {}
+
+
+def test_csrf_absent_when_cookie_missing():
+    assert rw.derive_csrf_headers("https://x.com/foo", "auth_token=z") == {}
+
+
+# --- receipt parse ----------------------------------------------------------
+def test_receipt_prefers_permalink_then_ids():
+    assert rw.receipt_from_json({"json": {"data": {"permalink": "/r/x/c/1", "id": "t1_9"}}}) == "/r/x/c/1"
+    assert rw.receipt_from_json({"data": {"create_tweet": {"tweet_results": {"rest_id": "1899"}}}}) == "1899"
+    assert rw.receipt_from_json({"nothing": True}) == ""
+
+
+def test_outcome_2xx_is_ok_with_receipt():
+    out = rw.outcome_from_response(200, '{"id_str": "1899"}', 42)
+    assert out.ok is True and out.receipt == "1899" and out.status == 200
+
+
+def test_outcome_2xx_non_json_is_ok_generic_receipt():
+    out = rw.outcome_from_response(201, "created", 5)
+    assert out.ok is True and out.receipt == "ok"
+
+
+def test_outcome_4xx_is_error():
+    out = rw.outcome_from_response(403, "forbidden csrf", 9)
+    assert out.ok is False and "403" in out.error
+
+
+# --- end-to-end with the network + session stubbed --------------------------
+def test_full_replay_success(monkeypatch):
+    p_arm(monkeypatch)
+    monkeypatch.setattr(rw, "get_session", lambda d: ("ct0=tok; sess=z", "UA/1.0"))
+    seen = {}
+
+    def fake_issue(method, url, body, headers):
+        seen["method"], seen["url"], seen["body"], seen["headers"] = method, url, body, headers
+        return 200, '{"json": {"data": {"things": [{"data": {"name": "t1_new", "permalink": "/r/x/c/a/_/t1_new"}}]}}}'
+
+    monkeypatch.setattr(rw, "issue_request", fake_issue)
+    out = rw.replay_write("POST", "https://www.reddit.com/api/comment", {"text": "nice", "thing_id": "t3_a"},
+                          "https://www.reddit.com", p_reddit_route())
+    assert out.ok is True and out.receipt == "/r/x/c/a/_/t1_new"
+    assert seen["method"] == "POST" and "Cookie" in seen["headers"]
+    assert "ct0=tok" in seen["headers"]["Cookie"]        # live-borrowed session, not persisted
+
+
+def test_full_replay_no_session_is_typed_miss(monkeypatch):
+    p_arm(monkeypatch)
+    def boom(domain):
+        raise RuntimeError("Not logged in")
+    monkeypatch.setattr(rw, "get_session", boom)
+    out = rw.replay_write("POST", "https://www.reddit.com/api/comment", {"text": "hi"},
+                          "https://www.reddit.com", p_reddit_route())
+    assert out.ok is False and "no borrowable session" in out.error
+
+
+def test_full_replay_site_reject_is_typed_error(monkeypatch):
+    p_arm(monkeypatch)
+    monkeypatch.setattr(rw, "get_session", lambda d: ("sess=z", "UA/1.0"))
+    monkeypatch.setattr(rw, "issue_request", lambda *a: (429, "rate limited"))
+    out = rw.replay_write("POST", "https://www.reddit.com/api/comment", {"text": "hi"},
+                          "https://www.reddit.com", p_reddit_route())
+    assert out.ok is False and "429" in out.error

backend/tests/test_site_write_registry.py

@@ -98,3 +98,46 @@ async def test_tool_no_url_yet_is_an_error_not_a_crash():
 async def test_tool_missing_action_is_an_error():
     out = await BA.run_api_write({"text": "hi"}, "https://www.reddit.com/r/x/")
     assert "error" in out and "action" in out["error"].lower()
+
+
+# --- general capture-replay tier (action='route') ---------------------------
+@pytest.mark.asyncio
+async def test_registry_route_write_wraps_replay_outcome(monkeypatch):
+    from backend.apps.agents.browser import route_write as rw
+    monkeypatch.setattr(reg, "p_ensure_session_env", lambda: None)
+    monkeypatch.setattr(rw, "replay_write",
+                        lambda m, u, b, o, c: rw.ReplayOutcome(ok=True, receipt="t1_z", latency_ms=88))
+    res = await reg.api_route_write("https://www.reddit.com", "POST",
+                                    "https://www.reddit.com/api/comment", {"text": "hi"}, [])
+    assert res.ok is True and res.receipt == "t1_z"
+    assert res.domain == "www.reddit.com" and res.action == "route"
+
+
+@pytest.mark.asyncio
+async def test_tool_route_fetches_captured_and_replays(monkeypatch):
+    # The tool fetches the site's captured write routes (safety wall) then replays. Both boundaries
+    # (the renderer list + the replay) are stubbed; this proves the wiring shape end to end.
+    async def fake_exec(tool, params, bid, tid):
+        assert tool == "BrowserListRoutes" and params == {"writes": True}
+        return {"routes": [{"method": "POST", "template": "https://www.reddit.com/api/comment"}]}
+
+    async def fake_route_write(origin, method, url, body, captured):
+        assert origin == "https://www.reddit.com" and method == "POST"
+        assert len(captured) == 1 and captured[0].template.endswith("/api/comment")
+        return reg.WriteResult(ok=True, action="route", domain="www.reddit.com",
+                               receipt="/r/x/c/a", latency_ms=120)
+
+    monkeypatch.setattr(BA, "execute_browser_tool", fake_exec)
+    monkeypatch.setattr(reg, "api_route_write", fake_route_write)
+    out = await BA.run_api_write(
+        {"action": "route", "method": "POST", "url": "https://www.reddit.com/api/comment",
+         "body": {"thing_id": "t3_a", "text": "hi"}},
+        "https://www.reddit.com/r/x/comments/a/", "b1", "t1")
+    assert out.get("ok") is True and "/r/x/c/a" in out["text"]
+
+
+@pytest.mark.asyncio
+async def test_tool_route_needs_a_url():
+    out = await BA.run_api_write({"action": "route", "method": "POST"},
+                                 "https://www.reddit.com/r/x/", "b1", "t1")
+    assert "error" in out and "url" in out["error"].lower()

frontend/src/shared/browserCommandHandler.ts

@@ -1612,8 +1612,12 @@ async function handleDetectWebMCP(wv: BrowserWebview): Promise<Record<string, an
   }
 }
 
-// Tier 2: the safe GET routes captured for the current site, so the agent can fetch data directly instead of re-scraping the UI. Only same-origin GET/HEAD routes are listed; those are all that replay_route will run.
-async function handleListRoutes(wv: BrowserWebview): Promise<Record<string, any>> {
+// Tier 2: the API routes captured for the current site, so the agent can act directly instead of
+// re-scraping/clicking the UI. Default lists the safe GET/HEAD routes (all replay_route will run).
+// With { writes: true } it lists the MUTATING routes (POST/PUT/PATCH/DELETE) the site's own UI
+// fired, for BrowserApiWrite's general 'route' path; the write itself is same-origin + captured +
+// session-borrowed + flag-gated in the backend, this only SURFACES the endpoint shape.
+async function handleListRoutes(wv: BrowserWebview, params?: Record<string, any>): Promise<Record<string, any>> {
   const bridge = (window as any).openswarm?.cdpRoutesGet as
     | ((id: number, origin?: string) => Promise<any[]>) | undefined;
   if (!bridge) return { error: 'Route capture not available, restart the app.' };
@@ -1621,6 +1625,21 @@ async function handleListRoutes(wv: BrowserWebview): Promise<Record<string, any>
   try { origin = new URL(wv.getURL()).origin; } catch {}
   let routes: any[] = [];
   try { routes = (await bridge(wv.getWebContentsId(), origin)) || []; } catch {}
+
+  if (params?.writes) {
+    const writes = routes.filter((r) => r && r.safe === false);
+    if (!writes.length) {
+      return { text: 'No write (POST/PUT/PATCH/DELETE) API routes captured for this site yet. Do the write once through the UI so it gets recorded, then the route path can replay it.', url: wv.getURL() };
+    }
+    const wlines = writes.slice(0, 40).map((r) => `${r.method} ${r.template}  body-shape: ${JSON.stringify(r.bodyShape)} (seen ${r.hits}x)`);
+    return {
+      text: `Write endpoints this site's UI uses (for BrowserApiWrite action='route'). Pass the `
+        + `method + url + a body matching the shape, with your content in the text field:\n${wlines.join('\n')}`,
+      routes: writes.slice(0, 40),
+      url: wv.getURL(),
+    };
+  }
+
   const safe = routes.filter((r) => r && r.safe);
   if (!safe.length) {
     return { text: 'No replayable (GET) API routes captured for this site yet. Use the page first so they get recorded, then try again.', url: wv.getURL() };
@@ -1870,7 +1889,7 @@ async function runBrowserCommand(
         result = await handleDetectWebMCP(wv);
         break;
       case 'list_routes':
-        result = await handleListRoutes(wv);
+        result = await handleListRoutes(wv, params);
         break;
       case 'click_by_name':
         result = await handleClickByName(wv, params);