diff --git a/backend/apps/agents/browser/site_write_registry.py b/backend/apps/agents/browser/site_write_registry.py
new file mode 100644
index 00000000..cfcfc515
--- /dev/null
+++ b/backend/apps/agents/browser/site_write_registry.py
@@ -0,0 +1,123 @@
+"""The API-first write tier, unified for the browser agent.
+
+When a write targets a site that has a borrowed-session write adapter, route the write HERE
+instead of UI puppeteering: borrow the user's live cookies, call the site's OWN write API, and
+return the site's typed receipt (its own id / permalink = proof it landed). This is
+deterministic (a typed success/error envelope, no captcha on the API surface, no DOM selector to
+drift) and ~50-190x faster than driving the UI (measured Reddit: 271ms vs 13-52s). Adding a site
+is one adapter entry; a site with no adapter falls back to the existing UI+model write path.
+
+Live-validated end to end on Reddit (comment 271ms + reversible delete 246ms, typed receipts).
+"""
+
+import asyncio
+import os
+import time
+from typing import Any, Callable, Dict, FrozenSet, Tuple
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.reddit_mcp_shim import reddit_writes
+
+
+class WriteResult(BaseModel):
+    """The typed outcome of an API-first write. `receipt` is the site's own id/permalink, the
+    proof the write landed (a real receipt, not a pixel guess); `ok` is False with a legible
+    `error` when the site's API rejected it or no session could be borrowed."""
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    ok: bool
+    action: str
+    domain: str
+    receipt: str = ""
+    error: str = ""
+    latency_ms: int = 0
+
+
+@typechecked
+def p_reddit_dispatch(action: str, params: Dict[str, Any]) -> Dict[str, Any]:
+    """Route a generic write action to the proven reddit_writes function; returns its receipt
+    dict (raises RedditError on the API's own error envelope, surfaced as ok=False upstream)."""
+    if action in ("comment", "reply"):
+        return reddit_writes.comment(str(params["parent_id"]), str(params["text"]))
+    if action in ("post", "submit"):
+        return reddit_writes.submit(
+            str(params["subreddit"]), str(params["title"]), str(params.get("kind", "self")),
+            str(params.get("text", "")), str(params.get("url", "")),
+            bool(params.get("nsfw", False)), bool(params.get("spoiler", False)),
+            bool(params.get("send_replies", True)),
+        )
+    if action == "edit":
+        return reddit_writes.edit(str(params["thing_id"]), str(params["text"]))
+    if action == "delete":
+        return reddit_writes.delete(str(params["thing_id"]))
+    raise ValueError(f"reddit adapter has no action {action!r}")
+
+
+# domain -> (actions it can do via the site's own API, sync dispatch fn). A dynamic-key registry
+# keyed by domain; adding a site is one row. X/others plug in the same shape once their write API
+# (GraphQL queryId + ct0) is proven, replacing their current UI-driving shim.
+P_ADAPTERS: Dict[str, Tuple[FrozenSet[str], Callable[[str, Dict[str, Any]], Dict[str, Any]]]] = {
+    "reddit.com": (frozenset({"comment", "reply", "post", "submit", "edit", "delete"}), p_reddit_dispatch),
+}
+
+
+@typechecked
+def has_api_write(domain: str, action: str) -> bool:
+    """True when this domain has a deterministic API adapter for this write action, so the agent
+    should route around the UI puppeteer tier."""
+    entry = P_ADAPTERS.get(domain.lower().strip().lstrip("."))
+    return bool(entry and action in entry[0])
+
+
+@typechecked
+def receipt_str(receipt: Dict[str, Any]) -> str:
+    """Flatten a site receipt dict into the single most-proof-bearing string (permalink beats a
+    bare id) so callers get one legible confirmation without knowing each site's shape."""
+    for key in ("permalink", "url", "id"):
+        v = receipt.get(key)
+        if v:
+            return str(v)
+    return "ok"
+
+
+@typechecked
+def p_ensure_session_env() -> None:
+    """Point session_source at the running backend so the in-process agent can borrow cookies the
+    same token-gated way the subprocess shims do (module globals are read at import, so patch
+    them). No-op once set."""
+    from backend.apps.social_shims import session_source as ss
+    port = os.environ.get("OPENSWARM_PORT", "8324")
+    ss.BACKEND_PORT = port
+    ss.BRIDGE_URL = f"http://127.0.0.1:{port}/api/browser-session/cookies"
+    if not ss.AUTH_TOKEN:
+        try:
+            from backend.auth import get_auth_token
+            ss.AUTH_TOKEN = get_auth_token() or ""
+        except Exception:
+            pass
+
+
+@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
+    failure (rejected by the site, no session, bad params) into a typed ok=False result rather
+    than raising, so the agent can fall back to the UI path on a miss without a crash."""
+    d = domain.lower().strip().lstrip(".")
+    entry = P_ADAPTERS.get(d)
+    if not entry or action not in entry[0]:
+        return WriteResult(ok=False, action=action, domain=d,
+                           error=f"no API-first adapter for {d}/{action}; use the UI path")
+    p_ensure_session_env()
+    dispatch = entry[1]
+    t0 = time.monotonic()
+    try:
+        receipt = await asyncio.to_thread(dispatch, action, params)
+        return WriteResult(ok=True, action=action, domain=d,
+                           receipt=receipt_str(receipt),
+                           latency_ms=int((time.monotonic() - t0) * 1000))
+    except Exception as e:
+        return WriteResult(ok=False, action=action, domain=d, error=str(e)[:200],
+                           latency_ms=int((time.monotonic() - t0) * 1000))
diff --git a/backend/tests/test_site_write_registry.py b/backend/tests/test_site_write_registry.py
new file mode 100644
index 00000000..53cf5ce8
--- /dev/null
+++ b/backend/tests/test_site_write_registry.py
@@ -0,0 +1,57 @@
+"""Unit tests for the API-first write registry: routing, receipt extraction, and the fail-safe
+(any adapter failure becomes a typed ok=False so the agent can fall back to the UI path, never a
+crash). Network is mocked; the live end-to-end proof is in PROTOCOL_apifirst.md."""
+import pytest
+
+from backend.apps.agents.browser import site_write_registry as reg
+
+
+def test_has_api_write_knows_reddit_and_rejects_unknown():
+    assert reg.has_api_write("reddit.com", "comment") is True
+    assert reg.has_api_write("REDDIT.COM", "delete") is True          # case + normalization
+    assert reg.has_api_write("reddit.com", "wire_money") is False     # unknown action
+    assert reg.has_api_write("example.com", "comment") is False       # no adapter
+
+
+def test_receipt_prefers_permalink_then_url_then_id():
+    assert reg.receipt_str({"permalink": "/r/x/c/1", "id": "t1_9"}) == "/r/x/c/1"
+    assert reg.receipt_str({"url": "https://x/p", "id": "t3_9"}) == "https://x/p"
+    assert reg.receipt_str({"id": "t1_9"}) == "t1_9"
+    assert reg.receipt_str({}) == "ok"
+
+
+@pytest.mark.asyncio
+async def test_api_write_routes_and_returns_typed_receipt(monkeypatch):
+    monkeypatch.setattr(reg, "p_ensure_session_env", lambda: None)
+    monkeypatch.setattr(reg.reddit_writes, "comment",
+                        lambda parent_id, text: {"id": "t1_abc", "permalink": "/r/test/comments/x/_/t1_abc"})
+    r = await reg.api_write("reddit.com", "comment", {"parent_id": "t3_x", "text": "hi"})
+    assert r.ok is True
+    assert r.receipt == "/r/test/comments/x/_/t1_abc"
+    assert r.action == "comment" and r.domain == "reddit.com"
+
+
+@pytest.mark.asyncio
+async def test_api_write_unknown_domain_is_typed_miss_not_crash(monkeypatch):
+    r = await reg.api_write("nosuchsite.com", "comment", {"text": "hi"})
+    assert r.ok is False
+    assert "no API-first adapter" in r.error
+
+
+@pytest.mark.asyncio
+async def test_api_write_adapter_failure_is_caught_as_typed_error(monkeypatch):
+    monkeypatch.setattr(reg, "p_ensure_session_env", lambda: None)
+    def boom(parent_id, text):
+        raise reg.reddit_writes.RedditError("RATELIMIT: try later")
+    monkeypatch.setattr(reg.reddit_writes, "comment", boom)
+    r = await reg.api_write("reddit.com", "comment", {"parent_id": "t3_x", "text": "hi"})
+    assert r.ok is False
+    assert "RATELIMIT" in r.error          # site's own error surfaced, no crash
+
+
+@pytest.mark.asyncio
+async def test_api_write_missing_required_param_is_typed_error(monkeypatch):
+    monkeypatch.setattr(reg, "p_ensure_session_env", lambda: None)
+    monkeypatch.setattr(reg.reddit_writes, "comment", lambda parent_id, text: {"id": "t1_x"})
+    r = await reg.api_write("reddit.com", "comment", {"text": "no parent id"})  # missing parent_id
+    assert r.ok is False and r.error            # KeyError -> typed miss, not a crash