[eric] x-mcp: convert to browser-delegation (X signs every request; drive the real logged-in card via data-testid DOM)

backend/apps/x_mcp_shim/handlers.py

@@ -3,10 +3,10 @@
 import json
 from typing import Any, Dict
 
+from backend.apps.social_shims.browser_action import BrowserActionError
 from backend.apps.social_shims.session_source import SessionUnavailable
 from backend.apps.x_mcp_shim import x_reads as reads
 from backend.apps.x_mcp_shim import x_writes as writes
-from backend.apps.x_mcp_shim.x_http import XError
 
 
 def mcp_ok(payload: Any) -> Dict[str, Any]:
@@ -24,7 +24,7 @@ def handle_tool_call(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
         return mcp_ok(p_dispatch(name, args))
     except SessionUnavailable as e:
         return mcp_err(str(e))
-    except XError as e:
+    except BrowserActionError as e:
         return mcp_err(str(e))
     except Exception as e:
         return mcp_err(f"x shim error: {e!r}")
@@ -34,19 +34,19 @@ def p_dispatch(name: str, a: Dict[str, Any]) -> Any:
     if name == "x_whoami":
         return reads.whoami()
     if name == "x_timeline":
-        return reads.timeline(a.get("kind", "foryou"), p_lim(a.get("count"), 20), a.get("cursor", ""))
+        return reads.timeline(a.get("kind", "foryou"), p_lim(a.get("count"), 20))
     if name == "x_user_tweets":
-        return reads.user_tweets(a.get("username", ""), p_lim(a.get("count"), 20), a.get("cursor", ""))
+        return reads.user_tweets(a.get("username", ""), p_lim(a.get("count"), 20))
     if name == "x_get_tweet":
         return reads.get_tweet(a.get("target", ""), p_lim(a.get("replies_limit"), 30))
     if name == "x_search":
-        return reads.search(a.get("query", ""), a.get("product", "top"), p_lim(a.get("count"), 20), a.get("cursor", ""))
+        return reads.search(a.get("query", ""), a.get("product", "top"), p_lim(a.get("count"), 20))
     if name == "x_get_user":
         return reads.get_user(a.get("username", ""))
     if name == "x_bookmarks":
-        return reads.bookmarks(p_lim(a.get("count"), 20), a.get("cursor", ""))
+        return reads.bookmarks(p_lim(a.get("count"), 20))
     if name == "x_notifications":
-        return reads.notifications(p_lim(a.get("count"), 20), a.get("cursor", ""))
+        return reads.notifications(p_lim(a.get("count"), 20))
     if name == "x_tweet":
         return writes.tweet(a.get("text", ""), a.get("reply_to", ""), a.get("quote_id", ""))
     if name == "x_delete_tweet":
@@ -61,7 +61,7 @@ def p_dispatch(name: str, a: Dict[str, Any]) -> Any:
         return writes.follow(a.get("username", ""), bool(a.get("unfollow")))
     if name == "x_send_dm":
         return writes.send_dm(a.get("recipient", ""), a.get("text", ""))
-    raise XError(f"Unknown tool: {name}")
+    raise BrowserActionError(f"Unknown tool: {name}")
 
 
 def p_lim(v: Any, default: int) -> int:

backend/apps/x_mcp_shim/rate_limit.py

@@ -1,33 +0,0 @@
-"""X's per-action pacing config on top of the shared RateLimiter.
-
-Reads are generous; posting is deliberately slow, likes/retweets moderate, follows
-and DMs slow, so the account never bursts like a bot. The shared core owns the
-algorithm + honors X's x-rate-limit-* headers and 429 backoff.
-"""
-
-from typing import Dict, Tuple
-
-from backend.apps.social_shims.rate_limit_core import RateLimiter
-
-# action -> (bucket_capacity, seconds_to_refill_one_token).
-BUCKETS: Dict[str, Tuple[float, float]] = {
-    "read": (30.0, 1.0),
-    "tweet": (3.0, 60.0),
-    "like": (20.0, 2.0),
-    "follow": (8.0, 6.0),
-    "dm": (5.0, 20.0),
-}
-
-p_limiter = RateLimiter(BUCKETS, min_gap_s=1.0, jitter_s=0.7)
-
-
-def acquire(action: str) -> None:
-    p_limiter.acquire(action)
-
-
-def note_response(status: int, headers: Dict[str, str]) -> None:
-    p_limiter.note_response(status, headers)
-
-
-def bucket_for(action: str) -> str:
-    return p_limiter.bucket_for(action)

backend/apps/x_mcp_shim/x_dom.py

@@ -0,0 +1,141 @@
+"""The JS that drives x.com's DOM inside the user's own logged-in card.
+
+X signs every API request with a browser-JS header we can't forge, so instead of calling
+the API we drive the real card: navigate + run these snippets via the perform_action bridge.
+This is the one brittle layer, X's data-testid attributes are the isolated assumption; if X
+reshuffles them, only this file changes. Selectors are the stable data-testid ones the web
+app itself uses (tweet, tweetText, like, reply, tweetButton, ...).
+"""
+
+import json
+from typing import List
+
+CAP_DEFAULT = 20
+
+
+def scrape_tweets_js(cap: int = CAP_DEFAULT) -> str:
+    """Poll for rendered tweets, then return a compact array of {id,author,text,likes,replies,url}."""
+    n = json.dumps(cap)
+    return (
+        "(async()=>{const cap=" + n + ";const sleep=ms=>new Promise(r=>setTimeout(r,ms));"
+        "const dl=Date.now()+7000;let arts=[];"
+        "while(Date.now()<dl){arts=[...document.querySelectorAll('article[data-testid=\"tweet\"]')];"
+        "if(arts.length)break;await sleep(400);}"
+        "const num=s=>{if(!s)return null;const m=(s+'').replace(/,/g,'').match(/(\\d+(?:\\.\\d+)?)([KM]?)/);"
+        "if(!m)return null;let v=parseFloat(m[1]);if(m[2]==='K')v*=1e3;if(m[2]==='M')v*=1e6;return Math.round(v);};"
+        "const out=[];const seen=new Set();"
+        "for(const a of arts){if(out.length>=cap)break;"
+        "const t=a.querySelector('[data-testid=\"tweetText\"]');const text=t?t.innerText:'';"
+        "let handle='';for(const l of a.querySelectorAll('a[href^=\"/\"]')){const m=(l.getAttribute('href')||'').match(/^\\/([A-Za-z0-9_]{1,15})$/);if(m){handle=m[1];break;}}"
+        "let url='',id='';const sl=a.querySelector('a[href*=\"/status/\"]');"
+        "if(sl){url='https://x.com'+sl.getAttribute('href').split('/photo')[0];const m=url.match(/status\\/(\\d+)/);if(m)id=m[1];}"
+        "if(id&&seen.has(id))continue;if(id)seen.add(id);"
+        "const lk=a.querySelector('[data-testid=\"like\"],[data-testid=\"unlike\"]');"
+        "const rp=a.querySelector('[data-testid=\"reply\"]');"
+        "out.push({id,author:handle,text:(text||'').slice(0,500),"
+        "likes:lk?num(lk.getAttribute('aria-label')):null,"
+        "replies:rp?num(rp.getAttribute('aria-label')):null,url});}"
+        "return out;})()"
+    )
+
+
+def whoami_js() -> str:
+    return (
+        "(()=>{const a=document.querySelector('[data-testid=\"AppTabBar_Profile_Link\"]');"
+        "const h=a?(a.getAttribute('href')||'').replace('/',''):'';"
+        "return{handle:h,logged_in:!!h};})()"
+    )
+
+
+def click_action_js(testids: List[str], done_testid: str, label: str) -> str:
+    """Click the first matching action button (like/retweet/follow); done_testid = the toggled state that means success/already."""
+    ids = json.dumps(testids)
+    done = json.dumps(done_testid)
+    lbl = json.dumps(label)
+    return (
+        "(async()=>{const ids=" + ids + ";const done=" + done + ";const label=" + lbl + ";"
+        "const sleep=ms=>new Promise(r=>setTimeout(r,ms));const dl=Date.now()+6000;"
+        "const find=()=>{for(const t of ids){const e=document.querySelector('[data-testid=\"'+t+'\"]');if(e)return e;}return null;};"
+        "let el=find();while(!el&&Date.now()<dl){await sleep(300);el=find();}"
+        "if(!el){if(document.querySelector('[data-testid=\"'+done+'\"]'))return{ok:true,already:true,action:label};"
+        "return{ok:false,error:label+' control not found'};}"
+        "el.scrollIntoView({block:'center'});el.click();await sleep(600);"
+        "return{ok:true,action:label};})()"
+    )
+
+
+def post_text_js(text: str, submit_testid: str = "tweetButton") -> str:
+    """Type into the focused/opened composer (Draft.js) and click submit. Used for compose + reply."""
+    t = json.dumps(text)
+    sub = json.dumps(submit_testid)
+    return (
+        "(async()=>{const text=" + t + ";const sub=" + sub + ";const sleep=ms=>new Promise(r=>setTimeout(r,ms));"
+        "const dl=Date.now()+7000;let box=null;"
+        "while(Date.now()<dl){box=document.querySelector('[data-testid=\"tweetTextarea_0\"]');if(box)break;await sleep(300);}"
+        "if(!box)return{ok:false,error:'composer not found'};"
+        "box.focus();document.execCommand('insertText',false,text);"
+        "box.dispatchEvent(new InputEvent('input',{bubbles:true,inputType:'insertText',data:text}));"
+        "await sleep(700);"
+        "let btn=document.querySelector('[data-testid=\"'+sub+'\"]');"
+        "if(!btn)return{ok:false,error:'submit button not found'};"
+        "if(btn.getAttribute('aria-disabled')==='true')return{ok:false,error:'submit disabled (empty/too long?)'};"
+        "btn.click();await sleep(1200);return{ok:true,posted:true};})()"
+    )
+
+
+def open_reply_js() -> str:
+    """On a tweet detail page, make sure the reply composer is open (click reply if the inline box isn't there)."""
+    return (
+        "(async()=>{const sleep=ms=>new Promise(r=>setTimeout(r,ms));"
+        "if(document.querySelector('[data-testid=\"tweetTextarea_0\"]'))return{ok:true,open:true};"
+        "const r=document.querySelector('[data-testid=\"reply\"]');"
+        "if(!r)return{ok:false,error:'reply button not found'};r.click();await sleep(1000);"
+        "return{ok:!!document.querySelector('[data-testid=\"tweetTextarea_0\"]'),open:true};})()"
+    )
+
+
+def profile_js() -> str:
+    return (
+        "(()=>{const nm=document.querySelector('[data-testid=\"UserName\"]');"
+        "const bio=document.querySelector('[data-testid=\"UserDescription\"]');"
+        "const grab=s=>{const a=document.querySelector('a[href$=\"/'+s+'\"]');return a?a.innerText.replace(/\\n/g,' '):null;};"
+        "const raw=nm?nm.innerText:'';const hm=raw.match(/@(\\w+)/);"
+        "return{name:raw.split('\\n')[0],handle:hm?hm[1]:'',bio:bio?bio.innerText:'',"
+        "following:grab('following'),followers:grab('verified_followers')||grab('followers')};})()"
+    )
+
+
+def retweet_js(undo: bool) -> str:
+    """Retweet is two clicks: the retweet button opens a menu, then confirm. Undo mirrors it."""
+    first = "unretweet" if undo else "retweet"
+    confirm = "unretweetConfirm" if undo else "retweetConfirm"
+    f = json.dumps(first)
+    c = json.dumps(confirm)
+    return (
+        "(async()=>{const first=" + f + ";const confirm=" + c + ";const sleep=ms=>new Promise(r=>setTimeout(r,ms));"
+        "const b=document.querySelector('[data-testid=\"'+first+'\"]');"
+        "if(!b)return{ok:false,error:first+' button not found (already done?)'};"
+        "b.scrollIntoView({block:'center'});b.click();await sleep(700);"
+        "const cf=document.querySelector('[data-testid=\"'+confirm+'\"]');"
+        "if(!cf)return{ok:false,error:'confirm menu item not found'};cf.click();await sleep(600);"
+        "return{ok:true,retweeted:!" + ("true" if undo else "false") + "};})()"
+    )
+
+
+def follow_js(unfollow: bool) -> str:
+    """The follow button's data-testid is '<userid>-follow' / '-unfollow'; match by suffix, fall back to button text."""
+    suffix = "-unfollow" if unfollow else "-follow"
+    label = "following" if unfollow else "follow"
+    s = json.dumps(suffix)
+    lbl = json.dumps(label)
+    return (
+        "(async()=>{const suf=" + s + ";const label=" + lbl + ";const sleep=ms=>new Promise(r=>setTimeout(r,ms));"
+        "const dl=Date.now()+6000;const find=()=>{let e=document.querySelector('[data-testid$=\"'+suf+'\"]');"
+        "if(e)return e;for(const b of document.querySelectorAll('[role=\"button\"]')){"
+        "if((b.textContent||'').trim().toLowerCase()===label)return b;}return null;};"
+        "let el=find();while(!el&&Date.now()<dl){await sleep(300);el=find();}"
+        "if(!el)return{ok:false,error:label+' button not found'};"
+        "el.scrollIntoView({block:'center'});el.click();await sleep(600);"
+        "if('" + ("true" if unfollow else "false") + "'==='true'){const c=document.querySelector('[data-testid=\"confirmationSheetConfirm\"]');if(c){c.click();await sleep(400);}}"
+        "return{ok:true,following:!" + ("true" if unfollow else "false") + "};})()"
+    )

backend/apps/x_mcp_shim/x_endpoints.py

@@ -1,75 +0,0 @@
-"""X (Twitter) web-client constants: the public bearer + the GraphQL operation map.
-
-The Authorization bearer below is the PUBLIC token x.com ships to every web client
-(logged-in or not); it is not a secret and not per-user. Real auth is the borrowed
-auth_token + ct0 cookies. The GraphQL queryIds drift whenever X redeploys its web
-app: refresh them by opening x.com in the OpenSwarm browser, watching the Network
-tab for /i/api/graphql/<id>/<OpName>, and pasting the new <id> here. This is the one
-drift-prone surface, deliberately isolated so a refresh is a one-line edit.
-
-Known gap: X's newest anti-automation header (x-client-transaction-id) is generated
-by obfuscated client JS we don't replicate; some endpoints may 404/403 without it.
-That's the X equivalent of Reddit's bearer-harvest assumption: structurally sound,
-not live-proven here.
-"""
-
-# Public web-app bearer (constant across all x.com web clients; not a credential).
-WEB_BEARER = (
-    "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D"
-    "1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA"
-)
-
-# OpName -> queryId. Drift-prone; see module docstring to refresh from a live capture.
-GRAPHQL_IDS = {
-    "UserByScreenName": "G3KGOASz96M-Qu0nwmGXNg",
-    "UserTweets": "E3opETHurmVJflFsUBVuUQ",
-    "TweetDetail": "xOhkmRac04YFZmOzU9PJHg",
-    "SearchTimeline": "nKAncKPF1fV1xltvF3UUlw",
-    "HomeTimeline": "uPv755D929tshj6KsxkSZg",
-    "HomeLatestTimeline": "8Rfm0g9b2-9La8Rmd1IPzw",
-    "Bookmarks": "j5KExFXxK0Nz1tQNXEx6KQ",
-    "CreateTweet": "znq5dRMnAYIRgIBQhGCRkg",
-    "DeleteTweet": "VaenaVgh5q5ih7kvyVjgtg",
-    "FavoriteTweet": "lI07N6Otwv1PhnEgXILM7A",
-    "UnfavoriteTweet": "ZYKSe-w7KEslx3JhSIk5LA",
-    "CreateRetweet": "ojPdsZsimiJrUGLR1sjUtA",
-    "DeleteRetweet": "iQtK4dl5hBmXewYZuEOKVw",
-    "CreateBookmark": "aoDbu3RHznuiSkQ9aNM67Q",
-    "DeleteBookmark": "Wlmlj2-xzyS1GN3a6cj-mQ",
-}
-
-# Feature flags X's GraphQL requires; a missing key 400s with "features cannot be null".
-# Also drift-prone; kept broad. Refresh alongside the queryIds.
-DEFAULT_FEATURES = {
-    "rweb_video_screen_enabled": False,
-    "profile_label_improvements_pcf_label_in_post_enabled": True,
-    "responsive_web_graphql_exclude_directive_enabled": True,
-    "verified_phone_label_enabled": False,
-    "creator_subscriptions_tweet_preview_api_enabled": True,
-    "responsive_web_graphql_timeline_navigation_enabled": True,
-    "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
-    "premium_content_api_read_enabled": False,
-    "communities_web_enable_tweet_community_results_fetch": True,
-    "c9s_tweet_anatomy_moderator_badge_enabled": True,
-    "responsive_web_grok_analyze_button_fetch_trends_enabled": False,
-    "responsive_web_grok_analyze_post_followups_enabled": True,
-    "responsive_web_jetfuel_frame": False,
-    "responsive_web_grok_share_attachment_enabled": True,
-    "articles_preview_enabled": True,
-    "responsive_web_edit_tweet_api_enabled": True,
-    "graphql_is_translatable_rweb_tweet_is_translatable_enabled": True,
-    "view_counts_everywhere_api_enabled": True,
-    "longform_notetweets_consumption_enabled": True,
-    "responsive_web_twitter_article_tweet_consumption_enabled": True,
-    "tweet_awards_web_tipping_enabled": False,
-    "responsive_web_grok_show_grok_translated_post": False,
-    "responsive_web_grok_analysis_button_from_backend": True,
-    "creator_subscriptions_quote_tweet_preview_enabled": False,
-    "freedom_of_speech_not_reach_fetch_enabled": True,
-    "standardized_nudges_misinfo": True,
-    "tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True,
-    "longform_notetweets_rich_text_read_enabled": True,
-    "longform_notetweets_inline_media_enabled": True,
-    "responsive_web_grok_image_annotation_enabled": True,
-    "responsive_web_enhance_cards_enabled": False,
-}

backend/apps/x_mcp_shim/x_http.py

@@ -1,104 +0,0 @@
-"""Low-level authed X (Twitter) transport.
-
-Borrow the user's x.com session (auth_token + ct0 cookies), attach the public web
-bearer + the ct0-derived CSRF header, and call x.com's own /i/api GraphQL + v1.1/v2
-surfaces exactly as the logged-in web client does. Rate-limited and self-refreshing
-on a 401/403 by re-borrowing the session. stdlib-only to match the sibling shims.
-"""
-
-import json
-import urllib.error
-import urllib.parse
-import urllib.request
-from typing import Any, Dict, Optional
-
-from backend.apps.social_shims.session_source import cookie_value, get_session, invalidate
-from backend.apps.x_mcp_shim import rate_limit
-from backend.apps.x_mcp_shim.x_endpoints import DEFAULT_FEATURES, GRAPHQL_IDS, WEB_BEARER
-
-DOMAIN = "x.com"
-API = "https://x.com/i/api"
-GRAPHQL = f"{API}/graphql"
-
-
-class XError(Exception):
-    """An X request failed in a way worth surfacing to the agent."""
-
-
-def p_send(method: str, url: str, *, data: Optional[bytes], content_type: Optional[str],
-           action: str, retried: bool = False) -> Any:
-    rate_limit.acquire(action)
-    cookie, ua = get_session(DOMAIN)
-    ct0 = cookie_value(DOMAIN, "ct0")
-    if not ct0:
-        invalidate(DOMAIN)
-        raise XError("No x.com CSRF cookie (ct0). Open x.com in the OpenSwarm browser, sign in, then retry.")
-    headers = {
-        "Authorization": f"Bearer {WEB_BEARER}",
-        "Cookie": cookie,
-        "User-Agent": ua,
-        "x-csrf-token": ct0,
-        "x-twitter-auth-type": "OAuth2Session",
-        "x-twitter-active-user": "yes",
-        "x-twitter-client-language": "en",
-        "Accept": "application/json",
-        "Referer": "https://x.com/",
-    }
-    if data is not None and content_type:
-        headers["Content-Type"] = content_type
-    req = urllib.request.Request(url, data=data, headers=headers, method=method)
-    try:
-        with urllib.request.urlopen(req, timeout=30.0) as resp:
-            status, raw, rhdr = resp.status, resp.read(), dict(resp.headers)
-    except urllib.error.HTTPError as e:
-        status, raw, rhdr = e.code, (e.read() if e.fp else b""), dict(e.headers or {})
-    except urllib.error.URLError as e:
-        raise XError(f"x.com unreachable: {getattr(e, 'reason', e)}")
-
-    rate_limit.note_response(status, {k.lower(): v for k, v in rhdr.items()})
-    if status in (401, 403) and not retried:
-        invalidate(DOMAIN)
-        return p_send(method, url, data=data, content_type=content_type, action=action, retried=True)
-    if status == 429:
-        raise XError("x.com is rate-limiting this account; slow down and retry shortly.")
-    if status >= 400:
-        raise XError(f"x.com HTTP {status}: {raw[:300].decode('utf-8', 'replace')}")
-    try:
-        return json.loads(raw.decode("utf-8", errors="replace") or "{}")
-    except json.JSONDecodeError:
-        return {"raw": raw.decode("utf-8", errors="replace")}
-
-
-def graphql(op: str, variables: Dict[str, Any], *, method: str = "GET",
-            features: bool = True, action: str = "read") -> Any:
-    """Call a GraphQL operation by name, looking up its (drift-prone) queryId."""
-    qid = GRAPHQL_IDS.get(op)
-    if not qid:
-        raise XError(f"Unknown GraphQL op {op!r}; add its queryId to x_endpoints.GRAPHQL_IDS.")
-    url = f"{GRAPHQL}/{qid}/{op}"
-    if method == "GET":
-        params = {"variables": json.dumps(variables, separators=(",", ":"))}
-        if features:
-            params["features"] = json.dumps(DEFAULT_FEATURES, separators=(",", ":"))
-        return p_send("GET", url + "?" + urllib.parse.urlencode(params),
-                      data=None, content_type=None, action=action)
-    body: Dict[str, Any] = {"variables": variables, "queryId": qid}
-    if features:
-        body["features"] = DEFAULT_FEATURES
-    return p_send("POST", url, data=json.dumps(body).encode(), content_type="application/json", action=action)
-
-
-def rest(method: str, path: str, *, params: Optional[Dict[str, Any]] = None,
-         form: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None,
-         action: str = "read") -> Any:
-    """Call a legacy v1.1/v2 endpoint (more stable than GraphQL for follow/DM). path includes the version."""
-    url = f"{API}/{path}"
-    if params:
-        url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
-    if json_body is not None:
-        return p_send(method, url, data=json.dumps(json_body).encode(),
-                      content_type="application/json", action=action)
-    if form is not None:
-        data = urllib.parse.urlencode({k: v for k, v in form.items() if v is not None}).encode()
-        return p_send(method, url, data=data, content_type="application/x-www-form-urlencoded", action=action)
-    return p_send(method, url, data=None, content_type=None, action=action)

backend/apps/x_mcp_shim/x_reads.py

@@ -1,191 +1,85 @@
-"""Read operations over x.com's own /i/api GraphQL + v1.1/v2 surfaces.
+"""Read operations for X, driven through the user's own logged-in x.com card.
 
-Returns compact, token-frugal tweet/user records (truncated text) instead of X's
-deeply-nested GraphQL firehose, so the agent sees what a human skims. The parsing
-walks for `tweet_results` nodes anywhere in the tree, which survives X's frequent
-timeline-shape reshuffles better than fixed index paths.
+X blocks pure-HTTP reads (it signs every request with browser-JS we can't forge), so we
+navigate the real card to the right URL, let it render, and scrape the DOM via the
+perform_action bridge. Free, undetectable, and immune to query-id/signature drift.
 """
 
 import re
-from typing import Any, Dict, List, Optional
+import urllib.parse
+from typing import Any, Dict, List
 
-from backend.apps.x_mcp_shim.x_http import XError, graphql, rest
+from backend.apps.social_shims.browser_action import last_json, perform
+from backend.apps.x_mcp_shim.x_dom import profile_js, scrape_tweets_js, whoami_js
 
-TEXT_CAP = 1200
-
-
-def p_trunc(s: Optional[str]) -> str:
-    s = s or ""
-    return s if len(s) <= TEXT_CAP else s[:TEXT_CAP] + f"... [+{len(s) - TEXT_CAP} chars]"
+DOMAIN = "x.com"
+SEARCH_F = {"latest": "live", "people": "user", "media": "media"}
 
 
 def tweet_id_of(target: str) -> str:
-    """Accept a status URL, a t-prefixed id, or a bare id; return the numeric id."""
     m = re.search(r"(\d{5,})", target or "")
     return m.group(1) if m else (target or "")
 
 
-def normalize_tweet(result: Any) -> Optional[Dict[str, Any]]:
-    if not isinstance(result, dict):
-        return None
-    if result.get("__typename") == "TweetWithVisibilityResults":
-        result = result.get("tweet", result)
-    legacy = result.get("legacy") or {}
-    if not legacy and not result.get("rest_id"):
-        return None
-    user_result = ((result.get("core") or {}).get("user_results") or {}).get("result") or {}
-    user_legacy = user_result.get("legacy") or {}
-    user_core = user_result.get("core") or {}
-    note = ((result.get("note_tweet") or {}).get("note_tweet_results") or {}).get("result") or {}
-    text = note.get("text") or legacy.get("full_text") or ""
-    return {
-        "id": result.get("rest_id") or legacy.get("id_str"),
-        "author": user_legacy.get("screen_name") or user_core.get("screen_name"),
-        "name": user_legacy.get("name") or user_core.get("name"),
-        "text": p_trunc(text),
-        "likes": legacy.get("favorite_count"),
-        "retweets": legacy.get("retweet_count"),
-        "replies": legacy.get("reply_count"),
-        "quotes": legacy.get("quote_count"),
-        "views": (result.get("views") or {}).get("count"),
-        "created_at": legacy.get("created_at"),
-        "lang": legacy.get("lang"),
-    }
-
-
-def p_collect_tweets(node: Any, out: List[Dict[str, Any]], cap: int) -> None:
-    if len(out) >= cap:
-        return
-    if isinstance(node, dict):
-        tr = node.get("tweet_results")
-        if isinstance(tr, dict) and isinstance(tr.get("result"), dict):
-            t = normalize_tweet(tr["result"])
-            if t and t.get("id") and not any(x["id"] == t["id"] for x in out):
-                out.append(t)
-        for v in node.values():
-            p_collect_tweets(v, out, cap)
-    elif isinstance(node, list):
-        for v in node:
-            p_collect_tweets(v, out, cap)
-
-
-def p_cursor(node: Any) -> Optional[str]:
-    found: List[str] = []
-
-    def walk(n: Any) -> None:
-        if found:
-            return
-        if isinstance(n, dict):
-            if n.get("cursorType") == "Bottom" and n.get("value"):
-                found.append(n["value"])
-            for v in n.values():
-                walk(v)
-        elif isinstance(n, list):
-            for v in n:
-                walk(v)
-
-    walk(node)
-    return found[0] if found else None
-
-
-def p_timeline_out(resp: Any, cap: int) -> Dict[str, Any]:
-    items: List[Dict[str, Any]] = []
-    p_collect_tweets(resp, items, cap)
-    return {"tweets": items, "cursor": p_cursor(resp)}
-
-
-def get_user(screen_name: str) -> Dict[str, Any]:
-    resp = graphql("UserByScreenName", {"screen_name": screen_name.lstrip("@")})
-    result = (((resp or {}).get("data") or {}).get("user") or {}).get("result") or {}
-    legacy = result.get("legacy") or {}
-    core = result.get("core") or {}
-    return {
-        "id": result.get("rest_id"),
-        "screen_name": legacy.get("screen_name") or core.get("screen_name"),
-        "name": legacy.get("name") or core.get("name"),
-        "bio": p_trunc(legacy.get("description")),
-        "followers": legacy.get("followers_count"),
-        "following": legacy.get("friends_count"),
-        "tweets": legacy.get("statuses_count"),
-        "verified": result.get("is_blue_verified") or legacy.get("verified"),
-        "created_at": legacy.get("created_at") or core.get("created_at"),
-    }
-
-
-def resolve_user_id(screen_name: str) -> str:
-    uid = get_user(screen_name).get("id")
-    if not uid:
-        raise XError(f"Could not resolve @{screen_name.lstrip('@')} to a user id.")
-    return str(uid)
+def p_tweets(url: str, cap: int, wait_ms: int = 3000) -> List[Dict[str, Any]]:
+    res = perform(DOMAIN, [
+        {"op": "navigate", "url": url},
+        {"op": "wait", "ms": wait_ms},
+        {"op": "evaluate", "expression": scrape_tweets_js(cap)},
+    ])
+    out = last_json(res)
+    return out if isinstance(out, list) else []
 
 
 def whoami() -> Dict[str, Any]:
-    settings = rest("GET", "1.1/account/settings.json")
-    screen = settings.get("screen_name", "")
-    out: Dict[str, Any] = {"screen_name": screen}
-    if screen:
-        try:
-            out["profile"] = get_user(screen)
-        except XError:
-            pass
-    return out
+    res = perform(DOMAIN, [
+        {"op": "navigate", "url": "https://x.com/home"},
+        {"op": "wait", "ms": 2200},
+        {"op": "evaluate", "expression": whoami_js()},
+    ])
+    return last_json(res)
 
 
-def timeline(kind: str, count: int, cursor: str) -> Dict[str, Any]:
-    op = "HomeLatestTimeline" if kind == "following" else "HomeTimeline"
-    variables: Dict[str, Any] = {"count": count, "includePromotedContent": False,
-                                 "latestControlAvailable": True, "withCommunity": True}
-    if cursor:
-        variables["cursor"] = cursor
-    return p_timeline_out(graphql(op, variables, method="POST"), count)
+def search(query: str, product: str, count: int) -> Dict[str, Any]:
+    q = urllib.parse.quote(query)
+    f = SEARCH_F.get((product or "top").lower())
+    url = f"https://x.com/search?q={q}&src=typed_query" + (f"&f={f}" if f else "")
+    tweets = p_tweets(url, count)
+    return {"query": query, "tweets": tweets, "count": len(tweets)}
 
 
-def user_tweets(screen_name: str, count: int, cursor: str) -> Dict[str, Any]:
-    uid = resolve_user_id(screen_name)
-    variables: Dict[str, Any] = {"userId": uid, "count": count, "includePromotedContent": False,
-                                 "withQuickPromoteEligibilityTweetFields": False, "withVoice": True,
-                                 "withV2Timeline": True}
-    if cursor:
-        variables["cursor"] = cursor
-    return p_timeline_out(graphql("UserTweets", variables), count)
+def timeline(kind: str, count: int) -> Dict[str, Any]:
+    tweets = p_tweets("https://x.com/home", count)
+    return {"kind": kind, "tweets": tweets, "count": len(tweets)}
 
 
-def get_tweet(target: str, count: int) -> Dict[str, Any]:
-    focal = tweet_id_of(target)
-    variables: Dict[str, Any] = {"focalTweetId": focal, "with_rux_injections": False,
-                                 "includePromotedContent": False, "withCommunity": True,
-                                 "withQuickPromoteEligibilityTweetFields": True, "withBirdwatchNotes": True,
-                                 "withVoice": True, "withV2Timeline": True}
-    out = p_timeline_out(graphql("TweetDetail", variables), count + 1)
-    tweets = out["tweets"]
-    main = next((t for t in tweets if t["id"] == focal), tweets[0] if tweets else {})
-    replies = [t for t in tweets if t.get("id") != main.get("id")]
-    return {"tweet": main, "replies": replies[:count]}
+def user_tweets(username: str, count: int) -> Dict[str, Any]:
+    h = username.lstrip("@")
+    tweets = p_tweets(f"https://x.com/{h}", count)
+    return {"username": h, "tweets": tweets, "count": len(tweets)}
 
 
-def search(query: str, product: str, count: int, cursor: str) -> Dict[str, Any]:
-    product = product.capitalize() if (product or "").lower() in ("top", "latest", "people", "media") else "Top"
-    variables: Dict[str, Any] = {"rawQuery": query, "count": count, "querySource": "typed_query", "product": product}
-    if cursor:
-        variables["cursor"] = cursor
-    return p_timeline_out(graphql("SearchTimeline", variables), count)
+def get_tweet(target: str, replies_limit: int) -> Dict[str, Any]:
+    url = target if str(target).startswith("http") else f"https://x.com/i/status/{tweet_id_of(target)}"
+    tweets = p_tweets(url, replies_limit + 1)
+    return {"tweet": tweets[0] if tweets else {}, "replies": tweets[1:replies_limit + 1]}
 
 
-def bookmarks(count: int, cursor: str) -> Dict[str, Any]:
-    variables: Dict[str, Any] = {"count": count, "includePromotedContent": False}
-    if cursor:
-        variables["cursor"] = cursor
-    return p_timeline_out(graphql("Bookmarks", variables), count)
+def get_user(username: str) -> Dict[str, Any]:
+    h = username.lstrip("@")
+    res = perform(DOMAIN, [
+        {"op": "navigate", "url": f"https://x.com/{h}"},
+        {"op": "wait", "ms": 2800},
+        {"op": "evaluate", "expression": profile_js()},
+    ])
+    return last_json(res)
 
 
-def notifications(count: int, cursor: str) -> Dict[str, Any]:
-    resp = rest("GET", "2/notifications/all.json", params={"count": count, "cursor": cursor or None})
-    notes = (resp or {}).get("globalObjects", {}).get("notifications", {})
-    out = []
-    for nid, n in list(notes.items())[:count]:
-        out.append({
-            "id": nid,
-            "text": (n.get("message") or {}).get("text"),
-            "timestamp_ms": n.get("timestampMs"),
-        })
-    return {"notifications": out}
+def bookmarks(count: int) -> Dict[str, Any]:
+    tweets = p_tweets("https://x.com/i/bookmarks", count)
+    return {"tweets": tweets, "count": len(tweets)}
+
+
+def notifications(count: int) -> Dict[str, Any]:
+    tweets = p_tweets("https://x.com/notifications", count)
+    return {"notifications": tweets, "count": len(tweets)}

backend/apps/x_mcp_shim/x_writes.py

@@ -1,74 +1,78 @@
-"""Write operations: everything a logged-in human does on X.
+"""Write operations for X, driven through the user's own logged-in x.com card.
 
-Tweets (with reply/quote), deletes, likes, retweets, bookmarks, follows, and DMs,
-all via the user's borrowed session. Each call rides the rate limiter's write buckets.
+Navigate the real card and click/type via the perform_action bridge, so X's own browser
+generates the request signature we can't forge from HTTP. Targets are tweet URLs from the
+read tools. tweet/reply/quote/like/retweet/follow are wired; DM stays a card-only action.
 """
 
 from typing import Any, Dict
 
-from backend.apps.x_mcp_shim.x_http import graphql, rest
-from backend.apps.x_mcp_shim.x_reads import normalize_tweet, resolve_user_id, tweet_id_of
+from backend.apps.social_shims.browser_action import BrowserActionError, last_json, perform
+from backend.apps.x_mcp_shim.x_dom import (
+    click_action_js,
+    follow_js,
+    open_reply_js,
+    post_text_js,
+    retweet_js,
+)
+from backend.apps.x_mcp_shim.x_reads import tweet_id_of
+
+DOMAIN = "x.com"
 
 
-def p_created_tweet(resp: Any) -> Dict[str, Any]:
-    result = ((((resp or {}).get("data") or {}).get("create_tweet") or {})
-              .get("tweet_results") or {}).get("result") or {}
-    t = normalize_tweet(result) or {}
-    return {"id": t.get("id"), "text": t.get("text")}
+def p_url(target: str) -> str:
+    return target if str(target).startswith("http") else f"https://x.com/i/status/{tweet_id_of(target)}"
 
 
 def tweet(text: str, reply_to: str, quote_id: str) -> Dict[str, Any]:
-    variables: Dict[str, Any] = {
-        "tweet_text": text,
-        "dark_request": False,
-        "media": {"media_entities": [], "possibly_sensitive": False},
-        "semantic_annotation_ids": [],
-    }
     if reply_to:
-        variables["reply"] = {"in_reply_to_tweet_id": tweet_id_of(reply_to), "exclude_reply_user_ids": []}
-    if quote_id:
-        variables["attachment_url"] = f"https://x.com/i/status/{tweet_id_of(quote_id)}"
-    return p_created_tweet(graphql("CreateTweet", variables, method="POST", action="tweet"))
-
-
-def delete_tweet(target: str) -> Dict[str, Any]:
-    tid = tweet_id_of(target)
-    graphql("DeleteTweet", {"tweet_id": tid, "dark_request": False}, method="POST", action="tweet")
-    return {"id": tid, "deleted": True}
+        steps = [
+            {"op": "navigate", "url": p_url(reply_to)},
+            {"op": "wait", "ms": 2800},
+            {"op": "evaluate", "expression": open_reply_js()},
+            {"op": "evaluate", "expression": post_text_js(text, "tweetButton")},
+        ]
+        return {"replied_to": reply_to, "result": last_json(perform(DOMAIN, steps))}
+    body = text if not quote_id else f"{text} {p_url(quote_id)}".strip()
+    steps = [
+        {"op": "navigate", "url": "https://x.com/compose/post"},
+        {"op": "wait", "ms": 2500},
+        {"op": "evaluate", "expression": post_text_js(body, "tweetButton")},
+    ]
+    return {"posted": True, "quote": bool(quote_id), "result": last_json(perform(DOMAIN, steps))}
 
 
 def like(target: str, unlike: bool) -> Dict[str, Any]:
-    tid = tweet_id_of(target)
-    graphql("UnfavoriteTweet" if unlike else "FavoriteTweet", {"tweet_id": tid}, method="POST", action="like")
-    return {"id": tid, "liked": not unlike}
+    js = click_action_js(["unlike"] if unlike else ["like"], "unlike" if not unlike else "like", "unlike" if unlike else "like")
+    steps = [{"op": "navigate", "url": p_url(target)}, {"op": "wait", "ms": 2600}, {"op": "evaluate", "expression": js}]
+    return {"target": target, "liked": not unlike, "result": last_json(perform(DOMAIN, steps))}
 
 
 def retweet(target: str, undo: bool) -> Dict[str, Any]:
-    tid = tweet_id_of(target)
-    if undo:
-        graphql("DeleteRetweet", {"source_tweet_id": tid, "dark_request": False}, method="POST", action="like")
-    else:
-        graphql("CreateRetweet", {"tweet_id": tid, "dark_request": False}, method="POST", action="like")
-    return {"id": tid, "retweeted": not undo}
+    steps = [{"op": "navigate", "url": p_url(target)}, {"op": "wait", "ms": 2600}, {"op": "evaluate", "expression": retweet_js(undo)}]
+    return {"target": target, "retweeted": not undo, "result": last_json(perform(DOMAIN, steps))}
 
 
-def bookmark(target: str, remove: bool) -> Dict[str, Any]:
-    tid = tweet_id_of(target)
-    graphql("DeleteBookmark" if remove else "CreateBookmark", {"tweet_id": tid}, method="POST", action="like")
-    return {"id": tid, "bookmarked": not remove}
+def follow(username: str, unfollow: bool) -> Dict[str, Any]:
+    h = username.lstrip("@")
+    steps = [{"op": "navigate", "url": f"https://x.com/{h}"}, {"op": "wait", "ms": 2600}, {"op": "evaluate", "expression": follow_js(unfollow)}]
+    return {"username": h, "following": not unfollow, "result": last_json(perform(DOMAIN, steps))}
 
 
-def follow(screen_name: str, unfollow: bool) -> Dict[str, Any]:
-    uid = resolve_user_id(screen_name)
-    path = "1.1/friendships/destroy.json" if unfollow else "1.1/friendships/create.json"
-    rest("POST", path, form={"user_id": uid}, action="follow")
-    return {"screen_name": screen_name.lstrip("@"), "following": not unfollow}
+def delete_tweet(target: str) -> Dict[str, Any]:
+    raise BrowserActionError(
+        f"Deleting a tweet needs the caret menu + a confirm dialog that's risky to click blind. "
+        f"Open {p_url(target)} in your X card and delete it there."
+    )
 
 
 def send_dm(recipient: str, text: str) -> Dict[str, Any]:
-    rid = recipient if recipient.isdigit() else resolve_user_id(recipient)
-    body = {"event": {"type": "message_create",
-                      "message_create": {"target": {"recipient_id": rid},
-                                         "message_data": {"text": text}}}}
-    rest("POST", "1.1/dm/new2.json", json_body=body, action="dm")
-    return {"to": recipient, "sent": True}
+    raise BrowserActionError(
+        f"DMs aren't wired for browser-driving yet. Open https://x.com/messages in your X card to DM {recipient!r}."
+    )
+
+
+def bookmark(target: str, remove: bool) -> Dict[str, Any]:
+    js = click_action_js(["removeBookmark"] if remove else ["bookmark"], "bookmark", "bookmark")
+    steps = [{"op": "navigate", "url": p_url(target)}, {"op": "wait", "ms": 2600}, {"op": "evaluate", "expression": js}]
+    return {"target": target, "bookmarked": not remove, "result": last_json(perform(DOMAIN, steps))}

backend/tests/test_x_mcp_shim.py

@@ -1,136 +1,112 @@
-"""Unit coverage for the X (Twitter) MCP shim's pure logic: tweet-id extraction, the
-deeply-nested GraphQL tweet/cursor walker, the rate limiter, and tool dispatch with
-the network mocked. Live posting can't be verified without a logged-in session, so the
-GraphQL contract is pinned here against canned x.com payload shapes."""
+"""Unit coverage for the X MCP shim (browser-delegation): tweet-id extraction and that each
+tool drives the right navigate + evaluate steps against the user's own x.com card. The
+perform_action bridge is mocked; live DOM behavior needs the running app + a logged-in X card
+(the data-testid selectors in x_dom are the isolated assumption, live-verified separately)."""
 
 import json
-import time
 
 from unittest.mock import patch
 
-from backend.apps.social_shims.session_source import SessionUnavailable
-from backend.apps.x_mcp_shim import rate_limit, x_reads, x_writes
+from backend.apps.social_shims.browser_action import BrowserActionError
+from backend.apps.x_mcp_shim import x_reads, x_writes
 from backend.apps.x_mcp_shim.handlers import handle_tool_call
-from backend.apps.x_mcp_shim.x_reads import normalize_tweet, tweet_id_of
+from backend.apps.x_mcp_shim.x_reads import tweet_id_of
+
+TWEET_URL = "https://x.com/alice/status/1850000000000000123"
 
 
 def p_text(result: dict) -> str:
     return result["content"][0]["text"]
 
 
-CANNED_TWEET = {
-    "__typename": "Tweet",
-    "rest_id": "111",
-    "legacy": {"full_text": "hello world", "favorite_count": 3, "retweet_count": 1,
-               "reply_count": 0, "id_str": "111", "lang": "en"},
-    "core": {"user_results": {"result": {"legacy": {"screen_name": "alice", "name": "Alice"}}}},
-    "views": {"count": "42"},
-}
+def perform_returning(payload):
+    """A fake perform() that records the steps and returns payload as the last evaluate's output."""
+    calls: dict = {}
 
-CANNED_TIMELINE = {"data": {"search_by_raw_query": {"search_timeline": {"timeline": {"instructions": [
-    {"type": "TimelineAddEntries", "entries": [
-        {"entryId": "tweet-111", "content": {"itemContent": {"tweet_results": {"result": CANNED_TWEET}}}},
-        {"entryId": "cursor-bottom", "content": {"cursorType": "Bottom", "value": "CURSOR123"}},
-    ]},
-]}}}}}
+    def fake(domain, steps):
+        calls["domain"] = domain
+        calls["ops"] = [s["op"] for s in steps]
+        calls["urls"] = [s.get("url") for s in steps if s["op"] == "navigate"]
+        return {"ok": True, "results": [{"text": json.dumps(payload)}]}
+
+    return fake, calls
 
 
-# -- id extraction + normalizers -------------------------------------------
+# -- id extraction ---------------------------------------------------------
 
-def test_tweet_id_from_url_and_bare():
-    assert tweet_id_of("https://x.com/alice/status/1850000000000000123") == "1850000000000000123"
+def test_tweet_id_of():
+    assert tweet_id_of(TWEET_URL) == "1850000000000000123"
     assert tweet_id_of("1850000000000000123") == "1850000000000000123"
-    assert tweet_id_of("t_nope") == "t_nope"
+    assert tweet_id_of("nope") == "nope"
 
 
-def test_normalize_tweet_normalizes():
-    t = normalize_tweet(CANNED_TWEET)
-    assert t["id"] == "111" and t["author"] == "alice" and t["name"] == "Alice"
-    assert t["text"] == "hello world" and t["likes"] == 3 and t["views"] == "42"
+# -- reads drive the right URL + scrape ------------------------------------
 
-
-def test_normalize_tweet_unwraps_visibility_wrapper():
-    wrapped = {"__typename": "TweetWithVisibilityResults", "tweet": CANNED_TWEET}
-    assert normalize_tweet(wrapped)["id"] == "111"
-
-
-def test_long_text_truncated():
-    big = {"rest_id": "9", "legacy": {"full_text": "x" * 4000, "id_str": "9"}}
-    body = normalize_tweet(big)["text"]
-    assert len(body) < 4000 and "+2800 chars" in body
-
-
-# -- rate limiter ----------------------------------------------------------
-
-def test_first_read_is_prompt():
-    start = time.time()
-    rate_limit.acquire("read")
-    assert time.time() - start < 1.2
-
-
-def test_429_backoff_delays_next_request():
-    rate_limit.note_response(429, {"retry-after": "1"})
-    start = time.time()
-    rate_limit.acquire("read")
-    assert time.time() - start >= 0.8
-
-
-# -- dispatch + normalizers (network mocked) -------------------------------
-
-def test_search_walks_nested_timeline():
-    with patch.object(x_reads, "graphql", return_value=CANNED_TIMELINE):
-        out = handle_tool_call("x_search", {"query": "openswarm", "count": 10})
+def test_search_drives_search_url_and_scrapes():
+    fake, calls = perform_returning([{"id": "111", "author": "bob", "text": "openswarm rocks", "likes": 42, "url": TWEET_URL}])
+    with patch.object(x_reads, "perform", fake):
+        out = handle_tool_call("x_search", {"query": "openswarm", "product": "top", "count": 5})
     data = json.loads(p_text(out))
     assert "isError" not in out
-    assert data["cursor"] == "CURSOR123"
-    assert data["tweets"][0]["id"] == "111"
-    assert data["tweets"][0]["author"] == "alice"
+    assert calls["domain"] == "x.com"
+    assert any("search?q=openswarm" in (u or "") for u in calls["urls"])
+    assert data["tweets"][0]["author"] == "bob" and data["count"] == 1
 
 
-def test_like_maps_to_favorite_op():
-    captured: dict = {}
+def test_user_tweets_navigates_to_profile():
+    fake, calls = perform_returning([])
+    with patch.object(x_reads, "perform", fake):
+        handle_tool_call("x_user_tweets", {"username": "@bob", "count": 5})
+    assert calls["urls"][0] == "https://x.com/bob"
 
-    def fake_graphql(op, variables, **kw):
-        captured["op"], captured["vars"] = op, variables
-        return {}
 
-    with patch.object(x_writes, "graphql", fake_graphql):
-        out = handle_tool_call("x_like", {"target": "https://x.com/a/status/1850000000000000111"})
-    assert captured["op"] == "FavoriteTweet"
-    assert captured["vars"]["tweet_id"] == "1850000000000000111"
+# -- writes drive navigate + evaluate on the card --------------------------
+
+def test_like_navigates_to_tweet_and_evaluates():
+    fake, calls = perform_returning({"ok": True, "action": "like"})
+    with patch.object(x_writes, "perform", fake):
+        out = handle_tool_call("x_like", {"target": TWEET_URL})
+    assert calls["urls"][0] == TWEET_URL
+    assert "evaluate" in calls["ops"]
     assert json.loads(p_text(out))["liked"] is True
 
 
-def test_unlike_maps_to_unfavorite_op():
-    captured: dict = {}
-    with patch.object(x_writes, "graphql", lambda op, v, **k: captured.setdefault("op", op) or {}):
-        handle_tool_call("x_like", {"target": "111", "unlike": True})
-    assert captured["op"] == "UnfavoriteTweet"
+def test_reply_opens_composer_then_posts():
+    fake, calls = perform_returning({"ok": True, "posted": True})
+    with patch.object(x_writes, "perform", fake):
+        out = handle_tool_call("x_tweet", {"text": "🔥", "reply_to": TWEET_URL})
+    assert calls["urls"][0] == TWEET_URL
+    assert calls["ops"].count("evaluate") == 2  # open_reply + post_text
+    assert json.loads(p_text(out))["replied_to"] == TWEET_URL
 
 
-def test_follow_resolves_id_and_hits_v11():
-    captured: dict = {}
+def test_compose_goes_to_composer():
+    fake, calls = perform_returning({"ok": True, "posted": True})
+    with patch.object(x_writes, "perform", fake):
+        handle_tool_call("x_tweet", {"text": "hello world"})
+    assert any("compose/post" in (u or "") for u in calls["urls"])
 
-    def fake_rest(method, path, **kw):
-        captured["path"], captured["form"] = path, kw.get("form")
-        return {}
 
-    with patch.object(x_writes, "resolve_user_id", return_value="999"), \
-            patch.object(x_writes, "rest", fake_rest):
+def test_follow_navigates_to_profile():
+    fake, calls = perform_returning({"ok": True, "following": True})
+    with patch.object(x_writes, "perform", fake):
         out = handle_tool_call("x_follow", {"username": "@bob"})
-    assert captured["path"] == "1.1/friendships/create.json"
-    assert captured["form"]["user_id"] == "999"
+    assert calls["urls"][0] == "https://x.com/bob"
     assert json.loads(p_text(out))["following"] is True
 
 
-def test_session_unavailable_is_actionable():
-    def boom(*a, **k):
-        raise SessionUnavailable("Not logged in to x.com. Open x.com in the OpenSwarm browser, sign in, then retry.")
+def test_delete_is_card_only():
+    out = handle_tool_call("x_delete_tweet", {"target": TWEET_URL})
+    assert out.get("isError") is True and "card" in p_text(out).lower()
 
-    with patch.object(x_reads, "rest", boom):
-        out = handle_tool_call("x_whoami", {})
-    assert out.get("isError") is True
-    assert "logged in" in p_text(out).lower()
+
+def test_no_card_error_surfaces():
+    def boom(*a, **k):
+        raise BrowserActionError("No x.com browser card is open. Open x.com in an OpenSwarm browser card and sign in, then retry.")
+
+    with patch.object(x_writes, "perform", boom):
+        out = handle_tool_call("x_like", {"target": TWEET_URL})
+    assert out.get("isError") is True and "card" in p_text(out).lower()
 
 
 def test_unknown_tool_errors():