Harden milestone sharing and X publishing

scripts/social-worker/worker.py

@@ -115,8 +115,35 @@ def _format_axis(value):
     return f"{value:g}" if value < 1000 else f"{value:,.0f}"
 
 
-def post_x(text, card):
-    oauth = OAuth1Session(required("X_API_KEY"), client_secret=required("X_API_SECRET"), resource_owner_key=required("X_ACCESS_TOKEN"), resource_owner_secret=required("X_ACCESS_TOKEN_SECRET"))
+def oauth_client():
+    return OAuth1Session(required("X_API_KEY"), client_secret=required("X_API_SECRET"), resource_owner_key=required("X_ACCESS_TOKEN"), resource_owner_secret=required("X_ACCESS_TOKEN_SECRET"))
+
+
+def find_existing_post(oauth, milestone):
+    """Reconcile an uncertain prior attempt before creating another X post."""
+    token = str(milestone.get("shareToken") or "").strip()
+    if not token:
+        raise RuntimeError("Milestone has no share token for duplicate-safe publishing")
+    identity = oauth.get("https://api.x.com/2/users/me", timeout=30)
+    identity.raise_for_status()
+    user_id = identity.json().get("data", {}).get("id")
+    if not user_id:
+        raise RuntimeError("X did not return the authenticated user id")
+    timeline = oauth.get(
+        f"https://api.x.com/2/users/{user_id}/tweets",
+        params={"max_results": 100, "tweet.fields": "created_at,entities", "exclude": "retweets,replies"},
+        timeout=30,
+    )
+    timeline.raise_for_status()
+    for post in timeline.json().get("data") or []:
+        urls = (post.get("entities") or {}).get("urls") or []
+        expanded = " ".join(str(url.get("expanded_url") or url.get("unwound_url") or "") for url in urls)
+        if token in expanded:
+            return post
+    return None
+
+
+def post_x(text, card, oauth):
     path = render_card(card) if card else None
     try:
         media_id = None
@@ -142,11 +169,13 @@ def run_once():
     if not milestone:
         return
     try:
-        result = post_x(milestone["text"], milestone.get("card"))
+        oauth = oauth_client()
+        existing = find_existing_post(oauth, milestone)
+        result = {"data": existing, "reconciled": True} if existing else post_x(milestone["text"], milestone.get("card"), oauth)
         tweet_id = result.get("data", {}).get("id")
         post_url = f"https://x.com/i/web/status/{tweet_id}" if tweet_id else None
         api("PATCH", base, {"id": milestone["id"], "result": "posted", "postUrl": post_url})
-        print(json.dumps({"posted": milestone["id"], "x": result}), flush=True)
+        print(json.dumps({"posted": milestone["id"], "reconciled": bool(existing), "x": result}), flush=True)
     except Exception as error:
         api("PATCH", base, {"id": milestone["id"], "result": "failed", "error": str(error)[:500]})
         print(f"Milestone post failed: {error}", flush=True)