fix(sources): twitter/youtube connector follow-ups (session config, rate limits, docs) (#180)

Follow-ups on the X/Twitter (#175) and YouTube (#177) connectors after running
both live, plus a shared cleanup.

Twitter:
- Session config (cookies/proxy/credentials dir) flows through Django settings
  (TWITTER_COOKIES_JSON / COOKIE_AUTH_TOKEN / COOKIE_CT0 / COOKIES_FILE /
  CREDENTIALS_DIR / PROXY), resolved per search call so a rotated cookie export
  applies without a restart; the documented-but-unwired JSON/file routes now
  work. CREDENTIALS_DIR defaults to an absolute path.
- Skip timestamp-less tweets instead of minting now() (which would poison the
  watermark and strand every older-but-new tweet).
- Honor X's rate-limit reset with an in-cycle retry loop (mirrors Reddit's 429
  loop; X's x-rate-limit-reset is an absolute epoch), ticking the poll-lease
  heartbeat through the wait.
- Watermark filter uses strict < not <= (same-second tweets were lost).
- Pin twikit to the locked rev.

Shared:
- Promote sleep_with_heartbeat and rate_limit_delay to connectors/base so
  Reddit and Twitter share one definition each.
- Rename ListenerError -> TwitterError (mirrors YouTubeError; drops the
  'Listener' name AGENTS.md reserves).

Docs/examples:
- Twitter starter (feed + watch); list twitter + youtube in the examples README.
- Per-connector credentials/ directory convention (gitignored) with a guide for
  generating the Twitter and YouTube cookie exports.
- README: product mentions in 'What it does', YouTube in the listeners + diagram,
  and the changelog-style section replaced by a credentials/changelog pointer.

Both connectors verified live end to end (feed -> watch -> judge). Full core
suite, ruff, and ty green; reviewed via /review-pr with findings addressed.

Author: joshpanka

README.md

@@ -240,25 +240,7 @@ Social listening is a crowded market (Brand24, Mention, Octolens, Syften, and to
 | Webhook methods | `POST`, `PUT`, `PATCH` |
 | Delivery audit | per-attempt `WatchActionDelivery` |
 
-## What we've done
-
-X/Twitter listening is the first connector added beyond the original Reddit / HN / RSS set. What shipped in this branch:
-
-- **`twitter_search` source kind** — a twikit-based connector that runs X search queries (mode Top/Latest, count) and maps results to a schema-parity `NewTweetPayload`, registered alongside the existing kinds with the same feed/watch/webhook pipeline.
-- **Live-cookie auth** — the connector authenticates with an existing X session cookie JSON (`TWITTER_COOKIES_JSON` → `auth_token`/`ct0`, a cookies file, or a login fallback), so no API key, proxy, or vendor API is required — the same live cookies the listening kit already used keep working.
-- **Reliability fixes from live polling** — a per-call twikit client (multi-source polls no longer crash with "Event loop is closed") and X's transient empty-body 404 mapped retryable instead of "tweet deleted", with a regression test. 587 tests green; all CI gates pass.
-- **Verified live end-to-end** — a real X poll through a feed → watch → webhook chain delivered 44/44 items with HTTP 200, payload matched field-for-field against the Twenty `socialEvent` intake contract (`item.handle → actorHandle`, `author → actorName`, `content → eventText`, `occurred_at → occurredAt`, `url → sourceUrl`, `key → dedupeKey`).
-
-YouTube listening followed via yt-dlp:
-
-- **`youtube_search` source kind** — a yt-dlp-based connector that runs YouTube search queries and maps results to a schema-parity `NewVideoPayload`, registered alongside the existing kinds with the same feed/watch/webhook pipeline.
-- **No authentication required** — public YouTube search works without credentials; optional cookie file for age-restricted content.
-- **Error taxonomy** — 5 error codes (`video_unavailable`, `rate_limited`, `js_runtime_missing`, `network_error`, `yt_dlp_error`) with retry semantics.
-- **Watermark-based deduplication** — videos newer than the source's `last_event_at` are surfaced.
-- **Metrics extraction** — views, likes, comments mapped from YouTube metadata.
-- **Thumbnail media** — full thumbnail URLs attached to payloads for rich display.
-
-Next up on the roadmap: **Facebook, TikTok, and Instagram connectors** (soon to be added), then Slack, LinkedIn, GitHub, Bluesky, and Mastodon.
+Two connectors ride an unofficial route and need a browser session cookie (X/Twitter always, YouTube only for age-restricted videos); see [apps/core/credentials/README.md](apps/core/credentials/README.md) for setup and the terms-of-service caveat. Per-release history is in the [changelog](CHANGELOG.md).
 
 ## Roadmap
 

apps/core/conf/settings/base.py

@@ -525,6 +525,26 @@
 # Google account. Empty (the default) disables it.
 YOUTUBE_COOKIES_FILE = os.environ.get("YOUTUBE_COOKIES_FILE", "")
 
+# X (Twitter) connector session config. The connector authenticates with an
+# existing x.com session, resolved per poll in priority order:
+#   TWITTER_COOKIES_JSON      a full JSON dict of x.com cookies (inline env)
+#   TWITTER_COOKIE_AUTH_TOKEN + TWITTER_COOKIE_CT0   the critical pair
+#   TWITTER_COOKIES_FILE      path to one cookie export (JSON dict or a
+#                             Get-cookies.txt-LOCALLY array)
+#   TWITTER_CREDENTIALS_DIR   dir of *.json cookie exports, each with an
+#                             optional <name>.proxy pin
+# Empty values fall through to the next route; all empty = guest mode (the
+# first search fails with a mapped `unauthorized`). See
+# apps/core/credentials/README.md for the on-disk convention.
+TWITTER_COOKIES_JSON = os.environ.get("TWITTER_COOKIES_JSON", "")
+TWITTER_COOKIE_AUTH_TOKEN = os.environ.get("TWITTER_COOKIE_AUTH_TOKEN", "")
+TWITTER_COOKIE_CT0 = os.environ.get("TWITTER_COOKIE_CT0", "")
+TWITTER_COOKIES_FILE = os.environ.get("TWITTER_COOKIES_FILE", "")
+TWITTER_CREDENTIALS_DIR = os.environ.get("TWITTER_CREDENTIALS_DIR", str(BASE_DIR / "credentials" / "twitter"))
+# Egress proxy for X requests (twikit passes it to httpx); a per-credential
+# <name>.proxy pin in TWITTER_CREDENTIALS_DIR overrides it.
+TWITTER_PROXY = os.environ.get("TWITTER_PROXY", "")
+
 # Product telemetry (anonymous, opt-out; see apps/core/telemetry + TELEMETRY.md).
 # POSTHOG_API_KEY defaults to the baked-in PUBLIC, WRITE-ONLY PostHog project key
 # (OpenMagpie's anonymous self-hosted project, PostHog Cloud US) so a self-hoster

apps/core/credentials/README.md

@@ -20,6 +20,58 @@ TWITTER_CREDENTIALS_DIR=/app/apps/core/credentials/twitter
 YOUTUBE_COOKIES_FILE=/app/apps/core/credentials/youtube/cookies.txt
 ```
 
+Settings are read per poll, so a refreshed export applies on the next cycle
+without a restart. `.env` changes themselves still need the usual
+`docker compose up -d --force-recreate`.
+
+## X / Twitter: generating a cookie export
+
+The `twitter_search` connector authenticates with an existing x.com browser
+session; there is no API key.
+
+1. Sign in to x.com in a browser.
+2. The minimal route needs just two cookies. In DevTools (Application ->
+   Cookies -> https://x.com), copy the values of `auth_token` and `ct0`
+   and set them directly:
+
+   ```
+   TWITTER_COOKIE_AUTH_TOKEN=<value>
+   TWITTER_COOKIE_CT0=<value>
+   ```
+
+3. For the file route instead, export the site's cookies with a browser
+   extension such as Cookie-Editor (export as JSON) or "Get cookies.txt
+   LOCALLY" (JSON export). Both shapes are accepted: a plain
+   `{name: value}` dict, or the extension's array of cookie objects. Save
+   it as `credentials/twitter/<name>.json`; the connector picks the first
+   usable export (sorted by filename) that carries the `auth_token`/`ct0`
+   pair.
+4. Optional: pin an egress proxy for one export by writing its URL to
+   `credentials/twitter/<name>.proxy` (same basename). `TWITTER_PROXY`
+   sets a global one.
+
+The full priority order (first configured route wins):
+`TWITTER_COOKIES_JSON` (inline JSON dict) -> the `auth_token`/`ct0` pair ->
+`TWITTER_COOKIES_FILE` (one export) -> `TWITTER_CREDENTIALS_DIR`.
+
+Sessions expire when X rotates them (or you log out in that browser);
+re-export and the next poll picks it up.
+
+## YouTube: generating cookies.txt (optional)
+
+Public YouTube search needs no credentials at all — set this up only if
+poll logs show relevant videos skipped as age-gated ("Sign in to confirm
+your age").
+
+1. Sign in to youtube.com, ideally in a private/incognito window (yt-dlp's
+   recommendation: export from a private session you then close, so the
+   browser doesn't rotate the exported cookies out from under you).
+2. Export the cookies in **Netscape format** with a "Get cookies.txt
+   LOCALLY"-style extension while on youtube.com (yt-dlp requires the
+   cookies.txt format here, not JSON).
+3. Save it as `credentials/youtube/cookies.txt` and set
+   `YOUTUBE_COOKIES_FILE` as above.
+
 It's recommended to use throwaway accounts for any cookies that land here:
 platforms flag and sometimes lock accounts whose sessions show up in
 automated traffic. These connectors use unofficial routes that may conflict

apps/core/pyproject.toml

@@ -24,7 +24,7 @@ dependencies = [
     "python-dotenv>=1.1",
     "pyyaml>=6.0",  # reads the examples/starters/*.yaml in seed_quickstart
     "trafilatura>=1.7",  # HTML -> readable article text for the engine's lazy external-link fetch
-    "twikit @ git+https://github.com/unclecode/twikit.git",  # X (Twitter) unofficial route (listeningkit-verified 2026 fork of d60/twikit)
+    "twikit @ git+https://github.com/unclecode/twikit.git@6a73ab97f4de09f79139f6308c9fb80029a9f5f7",  # X (Twitter) unofficial route (listeningkit-verified 2026 fork of d60/twikit), pinned: the fork's default branch floats and this code handles auth
     "yt-dlp>=2026.07.04",  # YouTube search connector (public API only)
     "ulid>=1.1",
 ]

apps/core/sources/connectors/base.py

@@ -103,6 +103,41 @@ def parse_rate_limit_wait(response: httpx.Response) -> float | None:
     return _as_positive_float(response.headers.get("X-RateLimit-Reset"))
 
 
+def rate_limit_delay(relative_wait: float | None, attempt: int, *, base: float, cap: float) -> float:
+    """Seconds to wait before retrying a rate-limited request: the caller's
+    header-derived relative wait when usable (finite, positive), else
+    exponential backoff `base * 2**attempt`. Capped at `cap` so a hostile or
+    far-future value can't stall a poll worker. Each connector converts its
+    own source shape to a relative wait first (Reddit: `parse_rate_limit_wait`'s
+    relative seconds; Twitter: an absolute `x-rate-limit-reset` epoch minus
+    now), keeping `base` / `cap` / retry-count as its own tunables."""
+    delay = relative_wait if relative_wait is not None and relative_wait > 0 else base * (2**attempt)
+    return min(delay, cap)
+
+
+# How often a backoff sleep ticks the caller's poll-lease heartbeat: long
+# enough not to thrash, short enough that a minute-scale wait renews the lease
+# several times over.
+HEARTBEAT_SLEEP_CHUNK_SECONDS = 15.0
+
+
+def sleep_with_heartbeat(total: float, heartbeat: Callable[[], bool] | None) -> None:
+    """Sleep `total` seconds, ticking `heartbeat` every chunk so the caller's
+    poll lease renews through the wait. The return value is deliberately
+    ignored (see the Connector.poll contract). No heartbeat (direct calls /
+    tests) = one plain sleep. Shared by every connector that backs off inside
+    poll() (Reddit's 429 retry, the twikit rate-limit wait)."""
+    if heartbeat is None:
+        time.sleep(total)
+        return
+    remaining = total
+    while remaining > 0:
+        chunk = min(remaining, HEARTBEAT_SLEEP_CHUNK_SECONDS)
+        time.sleep(chunk)
+        remaining -= chunk
+        heartbeat()
+
+
 # ── SSRF-safe fetch of the open web ───────────────────────────────────────
 # Two callers, one block POLICY (`common.ssrf` / `common.safe_http`):
 #   - RSS feeds (OPERATOR-chosen URLs): `validate_request_url`, an httpx request

apps/core/sources/connectors/reddit/connector.py

@@ -11,7 +11,14 @@
 from sources.payload_registry import register
 from sources.payloads import SourcePayload
 
-from ..base import BaseConnector, ConnectorParseError, parse_rate_limit_wait, read_response_capped
+from ..base import (
+    BaseConnector,
+    ConnectorParseError,
+    parse_rate_limit_wait,
+    rate_limit_delay,
+    read_response_capped,
+    sleep_with_heartbeat,
+)
 from .payloads import NewRedditPostPayload
 
 logger = logging.getLogger("sources")
@@ -70,40 +77,6 @@
 )
 RATE_LIMIT_DELAY_CAP_SECONDS = 60.0
 
-# Backoff sleeps tick the caller's `heartbeat` at this cadence so the poll
-# lease renews DURING the wait, not just between sources (the lease detects
-# dead holders; a deliberate wait is alive). Far inside the lease window
-# (POLL_LOCK_TIMEOUT_SECONDS, 600s), so even a worst-case stack of full
-# 60s waits never lets the lease lapse mid-source.
-HEARTBEAT_SLEEP_CHUNK_SECONDS = 15.0
-
-
-def _rate_limit_delay(header_wait: float | None, attempt: int) -> float:
-    """Seconds to wait before retrying a 429'd page: the wait the response's
-    rate-limit header asked for (`parse_rate_limit_wait`), else exponential in
-    the attempt number when no header was usable. Capped so a hostile / buggy
-    header can't stall a poll worker for minutes. (`parse_rate_limit_wait`
-    already screens NaN / inf / non-positive, so `header_wait` is a clean
-    positive float or None.)"""
-    delay = header_wait if header_wait is not None else RATE_LIMIT_BACKOFF_BASE_SECONDS * (2**attempt)
-    return min(delay, RATE_LIMIT_DELAY_CAP_SECONDS)
-
-
-def _sleep_with_heartbeat(total: float, heartbeat: Callable[[], bool] | None) -> None:
-    """Sleep `total` seconds, ticking `heartbeat` every chunk so the
-    caller's poll lease renews through the wait. The return value is
-    deliberately ignored (see the Connector.poll contract). No heartbeat
-    (direct calls / tests) = one plain sleep."""
-    if heartbeat is None:
-        time.sleep(total)
-        return
-    remaining = total
-    while remaining > 0:
-        chunk = min(remaining, HEARTBEAT_SLEEP_CHUNK_SECONDS)
-        time.sleep(chunk)
-        remaining -= chunk
-        heartbeat()
-
 
 def _entry_published(entry: Any) -> datetime | None:
     """feedparser exposes Atom `<published>` as `published_parsed`
@@ -188,7 +161,12 @@ def _get_page(
                     if attempt > 0:
                         logger.info("%s succeeded after %d retr%s", url, attempt, "y" if attempt == 1 else "ies")
                     return body
-                delay = _rate_limit_delay(parse_rate_limit_wait(response), attempt)
+                delay = rate_limit_delay(
+                    parse_rate_limit_wait(response),
+                    attempt,
+                    base=RATE_LIMIT_BACKOFF_BASE_SECONDS,
+                    cap=RATE_LIMIT_DELAY_CAP_SECONDS,
+                )
             # Sleep AFTER the `with` closes the 429 response, so the wait
             # never pins the streamed connection open.
             logger.info(
@@ -198,7 +176,7 @@ def _get_page(
                 attempt + 1,
                 MAX_RATE_LIMIT_RETRIES,
             )
-            _sleep_with_heartbeat(delay, heartbeat)
+            sleep_with_heartbeat(delay, heartbeat)
             attempt += 1
 
     def poll(

apps/core/sources/connectors/twitter/__init__.py

@@ -5,7 +5,7 @@
   - `client.py` ; `TwikitClient` wrapper (cookies env/file/credentials-dir,
     proxy attachment, error translation)
   - `payloads.py` ; `NewTweetPayload` (twikit Tweet -> SourcePayload)
-  - `errors.py` ; twikit error taxonomy -> canonical ListenerError
+  - `errors.py` ; twikit error taxonomy -> canonical TwitterError
 
 Future variants (user timeline, list timeline) reuse `TwikitClient` with
 their own spec + payload.