From 1dcf7317e55ca1c877aac320ec112bf4a14e97ef Mon Sep 17 00:00:00 2001
From: matthewdonsemail-lab <matthewdonsemail@gmail.com>
Date: Sat, 8 Aug 2026 17:50:21 +0700
Subject: [PATCH 1/6] feat(sources): add twitter_search connector
 (twikit-based) with schema parity

Ports the listeningkit Twitter-search listener into openmagpie-core as a built-in source kind, with twikit added as an openmagpie-core library dependency (git pin, unclecode fork) per operator direction - not vendored from listeningkit.

- schema: TwitterSearchSourceSpec (query, mode Top/Latest, count) added to the _BuiltinSourceSpec union; NewTweetPayload added to FeedItemData; schema.json regenerated (82 models); --check passes.
- connector: TwikitClient with the live-cookie credential chain (TWITTER_COOKIES_JSON -> TWITTER_COOKIE_AUTH_TOKEN/CT0 -> cookies file -> login fallback) so the working listeningkit cookie JSONs keep functioning; error taxonomy translated to ConnectorParseError; tests use fakes only (no live I/O).
- registration: twitter_search registered in sources/registry.py + connectors/__init__.py; payload self-registers; source-kind invariant test extended to the known five.
- infra: Dockerfile builder gains git for the git-pinned dep; docker-compose.override.yml (local-only, gitignored) publishes Postgres on host 5433 to coexist with the buzz stack.
- validation: 9 new connector tests + invariant tests green; CLI suite 145/145; ruff + ty clean.

Co-authored-by: Matthew Don <matthewdonsemail@gmail.com>
Signed-off-by: Matthew Don <matthewdonsemail@gmail.com>
---
 .gitignore                                    |   3 +
 apps/core/Dockerfile                          |   2 +-
 apps/core/feeds/tests_plugin_source_kinds.py  |   4 +-
 apps/core/pyproject.toml                      |   1 +
 apps/core/sources/connectors/__init__.py      |   2 +
 .../sources/connectors/twitter/__init__.py    |  17 ++
 .../core/sources/connectors/twitter/client.py | 176 +++++++++++++++++
 .../sources/connectors/twitter/connector.py   |  92 +++++++++
 .../core/sources/connectors/twitter/errors.py | 124 ++++++++++++
 .../sources/connectors/twitter/payloads.py    | 132 +++++++++++++
 apps/core/sources/registry.py                 |   2 +
 apps/core/sources/tests_twitter.py            | 136 +++++++++++++
 packages/openmagpie-schema/schema.json        | 178 +++++++++++++++++-
 .../src/openmagpie_schema/configs.py          |  45 ++++-
 .../src/openmagpie_schema/feed_payloads.py    |  25 ++-
 tools/schema_sync/models.py                   |   2 +
 uv.lock                                       | 109 +++++++++++
 17 files changed, 1044 insertions(+), 6 deletions(-)
 create mode 100644 apps/core/sources/connectors/twitter/__init__.py
 create mode 100644 apps/core/sources/connectors/twitter/client.py
 create mode 100644 apps/core/sources/connectors/twitter/connector.py
 create mode 100644 apps/core/sources/connectors/twitter/errors.py
 create mode 100644 apps/core/sources/connectors/twitter/payloads.py
 create mode 100644 apps/core/sources/tests_twitter.py

diff --git a/.gitignore b/.gitignore
index 928417ea..69043bce 100644
--- a/.gitignore
+++ b/.gitignore
@@ -81,3 +81,6 @@ cli/.venv/
 /*.csv
 /*.ndjson
 /*.jsonl
+
+# local-only: override docker-compose ports when the host ports are taken (see docker-compose.override.yml)
+docker-compose.override.yml
diff --git a/apps/core/Dockerfile b/apps/core/Dockerfile
index 8524f28a..1b9c677d 100644
--- a/apps/core/Dockerfile
+++ b/apps/core/Dockerfile
@@ -25,7 +25,7 @@ ENV PYTHONUNBUFFERED=1 \
 
 # Build deps for psycopg[c] (compiled against libpq). Builder-only.
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends build-essential libpq-dev \
+    && apt-get install -y --no-install-recommends build-essential git libpq-dev \
     && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app
diff --git a/apps/core/feeds/tests_plugin_source_kinds.py b/apps/core/feeds/tests_plugin_source_kinds.py
index a0f0b439..1252133d 100644
--- a/apps/core/feeds/tests_plugin_source_kinds.py
+++ b/apps/core/feeds/tests_plugin_source_kinds.py
@@ -26,6 +26,7 @@
     RedditSubredditSourceSpec,
     RssSourceSpec,
     SourceSpec,
+    TwitterSearchSourceSpec,
     _BuiltinSourceSpec,
     canonical_spec,
 )
@@ -308,7 +309,7 @@ def test_kind_literal_default_matches_source_kind(self) -> None:
         for m in members:
             self.assertEqual(m.model_fields["kind"].default, m.SOURCE_KIND, m.__name__)
 
-    def test_builtin_source_kinds_are_exactly_the_known_four(self) -> None:
+    def test_builtin_source_kinds_are_exactly_the_known_builtins(self) -> None:
         self.assertEqual(
             _BUILTIN_SOURCE_KINDS,
             frozenset(
@@ -317,6 +318,7 @@ def test_builtin_source_kinds_are_exactly_the_known_four(self) -> None:
                     RssSourceSpec.SOURCE_KIND,
                     HackerNewsFeedSourceSpec.SOURCE_KIND,
                     HackerNewsCommentSourceSpec.SOURCE_KIND,
+                    TwitterSearchSourceSpec.SOURCE_KIND,
                 }
             ),
         )
diff --git a/apps/core/pyproject.toml b/apps/core/pyproject.toml
index 39abc14a..739a7c6b 100644
--- a/apps/core/pyproject.toml
+++ b/apps/core/pyproject.toml
@@ -24,6 +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)
     "ulid>=1.1",
 ]
 
diff --git a/apps/core/sources/connectors/__init__.py b/apps/core/sources/connectors/__init__.py
index 041ba035..31939d1b 100644
--- a/apps/core/sources/connectors/__init__.py
+++ b/apps/core/sources/connectors/__init__.py
@@ -2,6 +2,7 @@
 from .hackernews import HackerNewsCommentConnector, HackerNewsFeedConnector
 from .reddit import RedditSubRedditConnector
 from .rss import RssConnector
+from .twitter import TwitterSearchConnector
 
 __all__ = [
     "Connector",
@@ -9,4 +10,5 @@
     "HackerNewsFeedConnector",
     "RedditSubRedditConnector",
     "RssConnector",
+    "TwitterSearchConnector",
 ]
diff --git a/apps/core/sources/connectors/twitter/__init__.py b/apps/core/sources/connectors/twitter/__init__.py
new file mode 100644
index 00000000..7643886c
--- /dev/null
+++ b/apps/core/sources/connectors/twitter/__init__.py
@@ -0,0 +1,17 @@
+"""X (Twitter) connector, unofficial route (twikit), ported from listeningkit.
+
+One file per concern:
+  - `connector.py` ; the `TwitterSearchConnector` impl (poll loop)
+  - `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
+
+Future variants (user timeline, list timeline) reuse `TwikitClient` with
+their own spec + payload.
+"""
+
+from .connector import TwitterSearchConnector
+from .payloads import NewTweetPayload
+
+__all__ = ["NewTweetPayload", "TwitterSearchConnector"]
diff --git a/apps/core/sources/connectors/twitter/client.py b/apps/core/sources/connectors/twitter/client.py
new file mode 100644
index 00000000..9905d01e
--- /dev/null
+++ b/apps/core/sources/connectors/twitter/client.py
@@ -0,0 +1,176 @@
+"""Twikit-based X (Twitter) client, unofficial route.
+
+Ported from REPOS/listeningkit/backend/packages/listeners/twitter/{client,
+config,credentials}.py so the openmagpie connector keeps the exact cookie
+JSON formats the live listeningkit setup uses (they work today; nothing
+about them is omitted or reshaped):
+
+- `TWITTER_COOKIES_JSON` (a full JSON dict of x.com cookies) — the live
+  `.env.local` source,
+- `TWITTER_COOKIE_AUTH_TOKEN` + `TWITTER_COOKIE_CT0` (the critical pair),
+- `TWITTER_COOKIES_FILE` (path to a twikit cookies.json),
+- `TWITTER_CREDENTIALS_DIR` (dir of `*.json` cookie exports, each with an
+  optional `*.proxy` pin), default `credentials/` relative to the core app.
+
+Proxy: `TWITTER_PROXY` env (or a per-credential `.proxy` pin in
+`TWITTER_CREDENTIALS_DIR`). twikit supports `proxy=` natively, so every
+request egresses through it (transport-level).
+
+twikit is async-only; the connector's `poll` is a sync iterator, so
+`TwikitClient.search` bridges with a fresh event loop per call
+(`asyncio.run`). One search per poll cycle is a bounded, short-lived
+loop; no shared loop state to leak across poll cycles.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Literal
+
+from twikit import Client
+from twikit.errors import TwitterException
+
+from .errors import ListenerError, map_bootstrap_failure, map_twikit_error
+
+# The product string twikit passes to X's search endpoint. The spec's
+# `latest` / `top` literals map 1:1 onto these (see connector.py).
+TwikitProduct = Literal["Latest", "Top"]
+
+log = logging.getLogger("sources.twitter")
+
+# The two cookies that make an authenticated twikit session work.
+CRITICAL_COOKIES = ("auth_token", "ct0")
+
+
+def _load_cookie_file(path: Path) -> dict[str, str]:
+    """Parse a Get-cookies.txt-LOCALLY JSON export (array) or a plain dict."""
+    data = json.loads(path.read_text(encoding="utf-8"))
+    if isinstance(data, dict):
+        return {str(k): str(v) for k, v in data.items() if v}
+    if isinstance(data, list):
+        out: dict[str, str] = {}
+        for item in data:
+            name, value = item.get("name"), item.get("value")
+            if name and value:
+                out[str(name)] = str(value)
+        return out
+    raise ValueError(f"{path}: JSON must be an object or an array of cookie objects")
+
+
+def load_cookies(
+    *,
+    cookies_json: str | None = None,
+    cookies_file: str | None = None,
+    credentials_dir: str | None = None,
+) -> tuple[dict[str, str], str | None]:
+    """Resolve the cookie dict + proxy for one X session, in priority order.
+
+    Returns ``(cookies, proxy)``. ``cookies`` is empty when nothing is
+    configured (the caller stays guest-mode; the first search then fails
+    with a mapped ``unauthorized`` error the poll loop handles). ``proxy``
+    comes from `TWITTER_PROXY` or a `<name>.proxy` pin next to the chosen
+    cookie export.
+    """
+    proxy = os.environ.get("TWITTER_PROXY", "").strip() or None
+
+    if cookies_json:
+        try:
+            data = json.loads(cookies_json)
+        except json.JSONDecodeError:
+            log.warning("TWITTER_COOKIES_JSON is not valid JSON; ignoring")
+        else:
+            if isinstance(data, dict) and data:
+                return {str(k): str(v) for k, v in data.items()}, proxy
+            log.warning("TWITTER_COOKIES_JSON is not a non-empty JSON object; ignoring")
+
+    individual = {name: os.environ.get(f"TWITTER_COOKIE_{name.upper()}", "").strip() for name in CRITICAL_COOKIES}
+    if all(individual.values()):
+        return individual, proxy
+
+    if cookies_file and Path(cookies_file).exists():
+        try:
+            return _load_cookie_file(Path(cookies_file)), proxy
+        except (json.JSONDecodeError, ValueError, OSError) as exc:
+            log.warning("TWITTER_COOKIES_FILE %s unreadable (%s); ignoring", cookies_file, exc)
+
+    directory = Path(credentials_dir or os.environ.get("TWITTER_CREDENTIALS_DIR", "credentials"))
+    if directory.exists():
+        for path in sorted(directory.glob("*.json")):
+            try:
+                cookies = _load_cookie_file(path)
+            except (json.JSONDecodeError, ValueError, OSError) as exc:
+                log.warning("skipping %s: %s", path.name, exc)
+                continue
+            if not {"auth_token", "ct0"}.issubset(cookies):
+                log.warning("skipping %s: missing auth_token/ct0 (critical pair)", path.name)
+                continue
+            proxy_path = path.with_suffix(".proxy")
+            pin = proxy_path.read_text().strip() if proxy_path.exists() else proxy
+            return cookies, pin
+        log.warning("credentials dir %s: no usable cookie set found", directory)
+    elif credentials_dir:
+        log.warning("credentials dir %s not found; no sessions loaded", credentials_dir)
+
+    return {}, proxy
+
+
+class ListenerErrorWrapper(Exception):
+    """Carries a canonical ListenerError through the pipeline."""
+
+    def __init__(self, err: ListenerError) -> None:
+        super().__init__(err.message)
+        self.error = err
+
+
+class TwikitClient:
+    """Thin, proxy-bound wrapper around the twikit async client."""
+
+    def __init__(
+        self,
+        *,
+        language: str = "en-US",
+        proxy: str | None = None,
+        user_agent: str | None = None,
+        cookies: dict[str, str] | None = None,
+        cookies_json: str | None = None,
+        cookies_file: str | None = None,
+        credentials_dir: str | None = None,
+    ) -> None:
+        self.proxy = proxy
+        if cookies is not None:
+            self._cookies = dict(cookies)
+        else:
+            self._cookies, self.proxy = load_cookies(
+                cookies_json=cookies_json,
+                cookies_file=cookies_file,
+                credentials_dir=credentials_dir,
+            )
+        self._client = Client(language=language, proxy=self.proxy, user_agent=user_agent)
+        self._ready = False
+
+    def _ensure_session(self) -> None:
+        if self._ready:
+            return
+        if self._cookies:
+            self._client.set_cookies(dict(self._cookies), clear_cookies=True)
+            log.info("session: loaded %d cookie(s)", len(self._cookies))
+        else:
+            log.warning("session: no cookies configured; guest mode only")
+        self._ready = True
+
+    async def _search_async(self, query: str, mode: TwikitProduct, count: int):
+        self._ensure_session()
+        try:
+            return await self._client.search_tweet(query, mode, count=count)
+        except TwitterException as exc:
+            raise ListenerErrorWrapper(map_twikit_error(exc, {"query": query, "mode": mode})) from exc
+        except Exception as exc:  # bootstrap failures (degraded shell) are not TwitterException
+            raise ListenerErrorWrapper(map_bootstrap_failure(exc, {"query": query})) from exc
+
+    def search(self, query: str, mode: TwikitProduct = "Latest", count: int = 20):
+        """Run one live search; returns a twikit Result[Tweet] (or a test double)."""
+        return asyncio.run(self._search_async(query, mode, count))
diff --git a/apps/core/sources/connectors/twitter/connector.py b/apps/core/sources/connectors/twitter/connector.py
new file mode 100644
index 00000000..ac4a1e4a
--- /dev/null
+++ b/apps/core/sources/connectors/twitter/connector.py
@@ -0,0 +1,92 @@
+"""X (Twitter) search connector, unofficial route (twikit).
+
+Polls a `twitter_search` source: one live X search per cycle via the
+twikit client (unclecode fork), mapping each result tweet to a
+`NewTweetPayload` newer than the source's `since` watermark.
+
+Error semantics follow the connector contract: any X/twikit failure is
+raised as `ConnectorParseError` (a `_RECOVERABLE_ERRORS` member at the
+poll seam), so a bad source logs + skips instead of aborting the feed
+cycle. The source's watermark stays put on failure, so the next cycle
+re-reads from the same point and the external_id dedup absorbs anything
+already recorded.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable, Iterator
+from datetime import datetime
+
+from openmagpie_schema.configs import TwitterSearchSourceSpec
+from sources.payload_registry import register
+from sources.payloads import SourcePayload
+
+from ..base import BaseConnector, ConnectorParseError
+from .client import ListenerErrorWrapper, TwikitClient, TwikitProduct
+from .payloads import NewTweetPayload
+
+log = logging.getLogger("sources.twitter")
+
+# Mode string twikit passes to X's search endpoint. The spec's `latest` /
+# `top` literals map 1:1 to twikit's `"Latest"` / `"Top"`.
+_TWIKIT_MODES: dict[str, TwikitProduct] = {"latest": "Latest", "top": "Top"}
+
+
+class TwitterSearchConnector(BaseConnector[TwitterSearchSourceSpec]):
+    """Polls one X (Twitter) search stream via the unofficial twikit route.
+
+    Live-mode semantics mirror the other connectors: every cycle yields
+    tweets newer than `since` (the Source row's `last_event_at`). There is
+    no pagination in phase 1: a search returns up to `spec.count` tweets
+    and the connector filters them by the watermark (X's own recency
+    ordering makes the first page the newest; a quiet stream needs no
+    backfill walk). Multi-account session rotation (listeningkit's
+    session_pool) is a follow-up; phase 1 uses one cookie set via the
+    client's env/file/credentials-dir resolution.
+    """
+
+    kind = TwitterSearchSourceSpec.SOURCE_KIND
+    payloads: list[type[SourcePayload]] = [NewTweetPayload]
+
+    # One stateless client; cookies + proxy resolved per search from the
+    # live env / credentials files (see client.load_cookies).
+    _client = TwikitClient()
+
+    def poll(
+        self,
+        spec: TwitterSearchSourceSpec,
+        since: datetime | None,
+        field_map: dict[str, str] | None = None,
+        heartbeat: Callable[[], bool] | None = None,
+    ) -> Iterator[SourcePayload]:
+        del field_map
+        del heartbeat
+        try:
+            results = self._client.search(spec.query, _TWIKIT_MODES[spec.mode], spec.count)
+        except ListenerErrorWrapper as exc:
+            err = exc.error
+            log.warning(
+                "twitter search failed query=%r code=%s retryable=%s: %s",
+                spec.query,
+                err.code,
+                err.retryable,
+                err.message,
+            )
+            raise ConnectorParseError(
+                f"twitter search {spec.display()} failed: {err.code}: {err.message} ({err.action})"
+            ) from exc
+
+        for tweet in results:
+            payload = NewTweetPayload.from_tweet(tweet, query=spec.query)
+            # Watermark filter: only surface tweets strictly newer than the
+            # cursor (the poll op advances the source watermark to the
+            # newest seen, so a tweet at the watermark is already recorded).
+            if since is not None and payload.occurred_at <= since:
+                continue
+            if spec.lang and payload.lang and payload.lang != spec.lang:
+                continue
+            yield payload
+
+
+register(TwitterSearchConnector.kind, TwitterSearchConnector.payloads)
diff --git a/apps/core/sources/connectors/twitter/errors.py b/apps/core/sources/connectors/twitter/errors.py
new file mode 100644
index 00000000..f3e7578e
--- /dev/null
+++ b/apps/core/sources/connectors/twitter/errors.py
@@ -0,0 +1,124 @@
+"""Error taxonomy for the X (Twitter) connector, ported from listeningkit.
+
+Every call into twikit can raise a ``TwitterException`` subclass (or a
+bootstrap failure when X serves a degraded shell). This module maps those
+to a canonical ``ListenerError``; the connector translates that into
+``ConnectorParseError`` at the poll boundary so the feed poll op recovers
+per-source (one bad source must not abort the feed cycle).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from twikit.errors import (
+    AccountLocked,
+    AccountSuspended,
+    BadRequest,
+    DuplicateTweet,
+    Forbidden,
+    InvalidMedia,
+    NotFound,
+    RequestTimeout,
+    ServerError,
+    TooManyRequests,
+    TweetNotAvailable,
+    TwitterException,
+    Unauthorized,
+    UserNotFound,
+    UserUnavailable,
+)
+
+TWIKIT_ERROR_CODE: dict[type[TwitterException], str] = {
+    BadRequest: "bad_request",
+    Unauthorized: "unauthorized",
+    Forbidden: "forbidden",
+    NotFound: "not_found",
+    RequestTimeout: "timeout",
+    TooManyRequests: "rate_limited",
+    ServerError: "upstream_error",
+    AccountSuspended: "account_suspended",
+    AccountLocked: "account_locked",
+    DuplicateTweet: "duplicate_tweet",
+    TweetNotAvailable: "tweet_unavailable",
+    InvalidMedia: "invalid_media",
+    UserNotFound: "user_not_found",
+    UserUnavailable: "user_unavailable",
+}
+
+
+@dataclass
+class ListenerError:
+    """Canonical error shape for one X fetch failure."""
+
+    code: str  # stable machine code, see TWIKIT_ERROR_CODE
+    message: str  # human-readable
+    retryable: bool  # safe to retry with backoff?
+    action: str  # what the ops layer should do
+    context: dict[str, Any] = field(default_factory=dict)
+    headers: dict[str, str] | None = None
+    rate_limit_reset: int | None = None  # unix ts from x-rate-limit-reset
+
+
+BOOTSTRAP_BLOCKED_MARKERS = (
+    "Couldn't get KEY_BYTE indices",
+    "Couldn't get key from the page source",
+)
+
+
+def map_bootstrap_failure(exc: Exception, context: dict[str, Any] | None = None) -> ListenerError:
+    """X served a degraded shell (bot wall) so twikit could not bootstrap its
+    ClientTransaction. Cause is almost always egress IP reputation; fix =
+    residential proxy (see listeningkit docs: proxy.md)."""
+    msg = str(exc)
+    code = "bootstrap_blocked" if any(m in msg for m in BOOTSTRAP_BLOCKED_MARKERS) else "internal"
+    action = (
+        "X served a degraded shell to this egress IP (no ondemand.s chunk map). "
+        "Use a residential proxy + browser fingerprint."
+        if code == "bootstrap_blocked"
+        else "unknown failure; log raw and retry with backoff"
+    )
+    return ListenerError(
+        code=code,
+        message=msg,
+        retryable=code == "internal",
+        action=action,
+        context=context or {},
+    )
+
+
+def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = None) -> ListenerError:
+    """Translate a twikit exception into a canonical ListenerError."""
+    code = TWIKIT_ERROR_CODE.get(type(exc), "twitter_error")
+    reset = getattr(exc, "rate_limit_reset", None)
+    headers = getattr(exc, "headers", None)
+
+    retryable_actions: dict[str, tuple[bool, str]] = {
+        "bad_request": (False, "fix query / payload; do not retry as-is"),
+        "unauthorized": (False, "refresh session (guest token / cookies) and retry once"),
+        "forbidden": (False, "rotate session + proxy pin; alert"),
+        "not_found": (False, "tweet/user no longer exists; skip"),
+        "timeout": (True, "retry with backoff"),
+        "rate_limited": (True, f"backoff until reset ({reset})"),
+        "upstream_error": (True, "retry with backoff; alert after 5 consecutive"),
+        "account_suspended": (False, "pause account mode; rotate to a different session; alert"),
+        "account_locked": (False, "Arkose challenge; pause account mode; alert"),
+        "duplicate_tweet": (False, "skip (dedupe by design)"),
+        "tweet_unavailable": (False, "skip"),
+        "invalid_media": (False, "skip"),
+        "user_not_found": (False, "skip"),
+        "user_unavailable": (False, "skip"),
+        "twitter_error": (True, "unknown upstream error; log raw and retry with backoff"),
+    }
+    retryable, action = retryable_actions.get(code, (True, "unknown; log and retry with backoff"))
+
+    return ListenerError(
+        code=code,
+        message=str(exc),
+        retryable=retryable,
+        action=action,
+        context=context or {},
+        headers=headers,
+        rate_limit_reset=reset,
+    )
diff --git a/apps/core/sources/connectors/twitter/payloads.py b/apps/core/sources/connectors/twitter/payloads.py
new file mode 100644
index 00000000..57df7602
--- /dev/null
+++ b/apps/core/sources/connectors/twitter/payloads.py
@@ -0,0 +1,132 @@
+"""X (Twitter) payloads: a tweet observed via the unofficial twikit route.
+
+Shapes the listeningkit SocialEvent normalization (see
+REPOS/listeningkit/docs/okf/backend/domains/twitter/unofficial/parsing.md)
+onto the openmagpie `SourcePayload` contract: the engine judges `title` +
+`content`, so a tweet's text goes to `content` and the author's handle
+becomes the within-kind `source_slug`. Metrics / refs / media stay on the
+payload as source-specific fields (available to actions that read them,
+omitted from the engine prompt unless included).
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any, ClassVar
+
+from openmagpie_schema.configs import TwitterSearchSourceSpec
+from sources.payloads import SourcePayload
+
+# Tweet URL base; a tweet's permalink is https://x.com/<handle>/status/<id>.
+X_STATUS_URL = "https://x.com"
+
+
+class NewTweetPayload(SourcePayload):
+    """A single tweet observed by a watched X search stream.
+
+    `author` is the user's display name; `handle` is the @screen_name and the
+    within-kind source slug (grouping items by producing account). `content`
+    is the tweet's full text (the engine's judgeable body). The rest is the
+    listeningkit event shape carried as payload fields: `metrics`, `refs`
+    (in_reply_to / quoted / retweet_of), `media`, `lang`.
+    """
+
+    PAYLOAD_KIND: ClassVar[str] = "new_tweet"
+
+    author: str = ""
+    handle: str = ""
+    lang: str = ""
+    metrics: dict[str, int | None] = {}
+    refs: dict[str, str | None] = {}
+    media: list[dict[str, Any]] = []
+
+    model_config = {"frozen": True, "extra": "ignore"}
+
+    def source_slug(self) -> str | None:
+        return self.handle or None
+
+    @classmethod
+    def sample(cls, variant: int = 0) -> NewTweetPayload:
+        n = variant + 1
+        tweet_id = str(999_000_000_000_000_000 + n)
+        handle = f"example_user_{n}"
+        return cls(
+            external_id=tweet_id,
+            kind=cls.PAYLOAD_KIND,
+            occurred_at=datetime(2026, 5, 27, 12, 0, tzinfo=UTC),
+            source=TwitterSearchSourceSpec.SOURCE_KIND,
+            title="",
+            content=f"Example tweet {n}: the post text that matched this watch.",
+            url=f"{X_STATUS_URL}/{handle}/status/{tweet_id}",
+            author=f"Example User {n}",
+            handle=handle,
+            lang="en",
+            metrics={"likes": 100 + n, "retweets": 20 + n, "replies": 5 + n, "quotes": 2 + n, "views": 1000 + n},
+            refs={"in_reply_to": None, "quoted": None, "retweet_of": None},
+            media=[],
+        )
+
+    @classmethod
+    def from_tweet(cls, tweet: Any, query: str | None = None) -> NewTweetPayload:
+        """Map a twikit `Tweet` (or a duck-typed test double) to a payload.
+
+        Kept attribute-driven (getattr with a default) so the connector's
+        unit tests can hand in lightweight fakes without importing twikit;
+        the real twikit Tweet supplies the same attributes. `query` is
+        recorded nowhere on the payload (the SourceSpec carries it); it is
+        accepted for symmetry with the listeningkit event's listenId and
+        future field_map use.
+        """
+        del query
+        tweet_id = str(getattr(tweet, "id", None) or "")
+        user = getattr(tweet, "user", None)
+        handle = ""
+        author = ""
+        if user is not None:
+            handle = str(getattr(user, "screen_name", None) or getattr(user, "username", None) or "")
+            author = str(getattr(user, "name", None) or "")
+        text = getattr(tweet, "full_text", None) or getattr(tweet, "text", None) or ""
+        created = getattr(tweet, "created_at_datetime", None) or getattr(tweet, "created_at", None)
+        occurred_at = created if isinstance(created, datetime) else datetime.now(UTC)
+        if occurred_at.tzinfo is None:
+            occurred_at = occurred_at.replace(tzinfo=UTC)
+        lang = str(getattr(tweet, "lang", None) or "")
+
+        def _id(obj: Any) -> str | None:
+            return str(getattr(obj, "id", None)) if obj is not None else None
+
+        media = []
+        for m in getattr(tweet, "media", None) or []:
+            media.append(
+                {
+                    "type": getattr(m, "type", None),
+                    "url": getattr(m, "media_url_https", None) or getattr(m, "media_url", None),
+                    "thumbnail": getattr(m, "thumbnail_url", None),
+                }
+            )
+
+        return cls(
+            external_id=tweet_id,
+            kind=cls.PAYLOAD_KIND,
+            occurred_at=occurred_at,
+            source=TwitterSearchSourceSpec.SOURCE_KIND,
+            title="",
+            content=text,
+            url=f"{X_STATUS_URL}/{handle}/status/{tweet_id}" if handle else "",
+            author=author,
+            handle=handle,
+            lang=lang,
+            metrics={
+                "likes": getattr(tweet, "favorite_count", None),
+                "retweets": getattr(tweet, "retweet_count", None),
+                "replies": getattr(tweet, "reply_count", None),
+                "quotes": getattr(tweet, "quote_count", None),
+                "views": getattr(tweet, "view_count", None),
+            },
+            refs={
+                "in_reply_to": _id(getattr(tweet, "in_reply_to", None)),
+                "quoted": _id(getattr(tweet, "quote", None)),
+                "retweet_of": _id(getattr(tweet, "retweeted_tweet", None)),
+            },
+            media=media,
+        )
diff --git a/apps/core/sources/registry.py b/apps/core/sources/registry.py
index bfa6713d..3ba421b9 100644
--- a/apps/core/sources/registry.py
+++ b/apps/core/sources/registry.py
@@ -15,6 +15,7 @@
     HackerNewsFeedConnector,
     RedditSubRedditConnector,
     RssConnector,
+    TwitterSearchConnector,
 )
 
 _REGISTRY: dict[str, Connector[Any]] = {
@@ -22,6 +23,7 @@
     RssConnector.kind: RssConnector(),
     HackerNewsFeedConnector.kind: HackerNewsFeedConnector(),
     HackerNewsCommentConnector.kind: HackerNewsCommentConnector(),
+    TwitterSearchConnector.kind: TwitterSearchConnector(),
 }
 
 # Core kinds captured before any plugin registers; a plugin can't replace one.
diff --git a/apps/core/sources/tests_twitter.py b/apps/core/sources/tests_twitter.py
new file mode 100644
index 00000000..217be08b
--- /dev/null
+++ b/apps/core/sources/tests_twitter.py
@@ -0,0 +1,136 @@
+"""Twitter search connector tests (offline, fake twikit results).
+
+The connector's only I/O is the twikit client (`TwikitClient.search`); these
+tests swap in a fake result iterator and pin: spec validation (the blank-query
+firehose guard), the watermark filter, the lang filter, error translation
+(ListenerErrorWrapper -> ConnectorParseError), and payload mapping (a duck-
+typed fake Tweet -> NewTweetPayload).
+"""
+
+from datetime import UTC, datetime
+from unittest import mock
+
+from django.test import SimpleTestCase
+from pydantic import ValidationError
+
+from openmagpie_schema.configs import TwitterSearchSourceSpec
+from sources.connectors.base import ConnectorParseError
+from sources.connectors.twitter.client import ListenerErrorWrapper
+from sources.connectors.twitter.connector import TwitterSearchConnector
+from sources.connectors.twitter.errors import ListenerError
+from sources.connectors.twitter.payloads import NewTweetPayload
+
+
+class _FakeUser:
+    def __init__(self, handle: str, name: str = "Some User"):
+        self.id = "987654321"
+        self.screen_name = handle
+        self.username = handle
+        self.name = name
+
+
+class _FakeTweet:
+    def __init__(
+        self,
+        tweet_id: str,
+        handle: str = "alice",
+        text: str = "hello from x",
+        created: datetime | None = None,
+        lang: str = "en",
+    ):
+        self.id = tweet_id
+        self.user = _FakeUser(handle)
+        self.full_text = text
+        self.created_at_datetime = created or datetime(2026, 6, 1, 12, 0, tzinfo=UTC)
+        self.created_at = self.created_at_datetime
+        self.lang = lang
+        self.favorite_count = 10
+        self.retweet_count = 2
+        self.reply_count = 1
+        self.quote_count = 0
+        self.view_count = 100
+        self.media = []
+        self.in_reply_to = None
+        self.quote = None
+        self.retweeted_tweet = None
+
+
+class TwitterSearchSourceSpecTests(SimpleTestCase):
+    def test_blank_query_rejected(self):
+        with self.assertRaises(ValidationError):
+            TwitterSearchSourceSpec(kind="twitter_search", query="   ")
+
+    def test_count_bounds(self):
+        with self.assertRaises(ValidationError):
+            TwitterSearchSourceSpec(kind="twitter_search", query="x", count=0)
+        with self.assertRaises(ValidationError):
+            TwitterSearchSourceSpec(kind="twitter_search", query="x", count=101)
+
+    def test_defaults(self):
+        spec = TwitterSearchSourceSpec(kind="twitter_search", query="social listening")
+        self.assertEqual(spec.mode, "latest")
+        self.assertEqual(spec.count, 20)
+        self.assertEqual(spec.lang, "")
+
+
+class TwitterSearchConnectorTests(SimpleTestCase):
+    def _connector(self, results):
+        client = mock.Mock()
+        client.search.return_value = results
+        conn = TwitterSearchConnector()
+        conn._client = client
+        return conn, client
+
+    def test_yields_payloads_newer_than_since(self):
+        spec = TwitterSearchSourceSpec(kind="twitter_search", query='"social listening"')
+        tweets = [
+            _FakeTweet("2", text="newer"),
+            _FakeTweet("1", text="older", created=datetime(2026, 5, 1, 12, 0, tzinfo=UTC)),
+        ]
+        conn, client = self._connector(tweets)
+        payloads = list(conn.poll(spec, since=datetime(2026, 5, 15, tzinfo=UTC)))
+        self.assertEqual(len(payloads), 1)
+        self.assertEqual(payloads[0].external_id, "2")
+        client.search.assert_called_once_with('"social listening"', "Latest", 20)
+
+    def test_lang_filter(self):
+        spec = TwitterSearchSourceSpec(kind="twitter_search", query="x", lang="es")
+        tweets = [_FakeTweet("1", lang="en"), _FakeTweet("2", lang="es")]
+        conn, _ = self._connector(tweets)
+        payloads = list(conn.poll(spec, since=None))
+        self.assertEqual([p.external_id for p in payloads], ["2"])
+
+    def test_error_maps_to_connector_parse_error(self):
+        spec = TwitterSearchSourceSpec(kind="twitter_search", query="x")
+        err = ListenerError(code="rate_limited", message="slow down", retryable=True, action="backoff")
+        client = mock.Mock()
+        client.search.side_effect = ListenerErrorWrapper(err)
+        conn = TwitterSearchConnector()
+        conn._client = client
+        with self.assertRaises(ConnectorParseError) as ctx:
+            list(conn.poll(spec, since=None))
+        self.assertIn("rate_limited", str(ctx.exception))
+
+    def test_mode_top_maps_to_twikit_top(self):
+        spec = TwitterSearchSourceSpec(kind="twitter_search", query="x", mode="top")
+        conn, client = self._connector([_FakeTweet("1")])
+        list(conn.poll(spec, since=None))
+        client.search.assert_called_once_with("x", "Top", 20)
+
+
+class NewTweetPayloadTests(SimpleTestCase):
+    def test_from_tweet(self):
+        p = NewTweetPayload.from_tweet(_FakeTweet("123", handle="alice", text="hi"))
+        self.assertEqual(p.external_id, "123")
+        self.assertEqual(p.handle, "alice")
+        self.assertEqual(p.content, "hi")
+        self.assertEqual(p.source, "twitter_search")
+        self.assertEqual(p.url, "https://x.com/alice/status/123")
+        self.assertEqual(p.metrics["likes"], 10)
+        self.assertEqual(p.refs["in_reply_to"], None)
+
+    def test_sample_distinct(self):
+        a = NewTweetPayload.sample(0)
+        b = NewTweetPayload.sample(1)
+        self.assertNotEqual(a.external_id, b.external_id)
+        self.assertEqual(a.PAYLOAD_KIND, "new_tweet")
diff --git a/packages/openmagpie-schema/schema.json b/packages/openmagpie-schema/schema.json
index 86183978..5960ac24 100644
--- a/packages/openmagpie-schema/schema.json
+++ b/packages/openmagpie-schema/schema.json
@@ -835,6 +835,9 @@
             {
               "$ref": "#/$defs/HackerNewsCommentPayload"
             },
+            {
+              "$ref": "#/$defs/NewTweetPayload"
+            },
             {
               "$ref": "#/$defs/FeedItemPayload"
             }
@@ -1733,6 +1736,124 @@
       "title": "NewRedditPostPayload",
       "type": "object"
     },
+    "NewTweetPayload": {
+      "additionalProperties": true,
+      "description": "`new_tweet`: one X (Twitter) tweet (TwitterSearchConnector, unofficial\ntwikit route). `content` is the tweet's full text (the engine's judgeable\nbody); `title` is empty (tweets have no headline). The listeningkit\nSocialEvent shape is carried as typed fields: `handle` (the @screen_name,\nalso the within-kind source slug), `author` (display name), `lang`,\n`metrics` (likes/retweets/replies/quotes/views), `refs`\n(in_reply_to / quoted / retweet_of), `media` (list of {type,url,thumbnail}).",
+      "properties": {
+        "author": {
+          "default": "",
+          "title": "Author",
+          "type": "string"
+        },
+        "content": {
+          "default": "",
+          "title": "Content",
+          "type": "string"
+        },
+        "external_id": {
+          "default": "",
+          "title": "External Id",
+          "type": "string"
+        },
+        "external_url": {
+          "default": "",
+          "title": "External Url",
+          "type": "string"
+        },
+        "handle": {
+          "default": "",
+          "title": "Handle",
+          "type": "string"
+        },
+        "kind": {
+          "const": "new_tweet",
+          "title": "Kind",
+          "type": "string"
+        },
+        "lang": {
+          "default": "",
+          "title": "Lang",
+          "type": "string"
+        },
+        "media": {
+          "default": [],
+          "items": {
+            "additionalProperties": true,
+            "type": "object"
+          },
+          "title": "Media",
+          "type": "array"
+        },
+        "metrics": {
+          "additionalProperties": {
+            "anyOf": [
+              {
+                "type": "integer"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          },
+          "default": {},
+          "title": "Metrics",
+          "type": "object"
+        },
+        "occurred_at": {
+          "anyOf": [
+            {
+              "format": "date-time",
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "title": "Occurred At"
+        },
+        "parent_external_id": {
+          "default": "",
+          "title": "Parent External Id",
+          "type": "string"
+        },
+        "refs": {
+          "additionalProperties": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          },
+          "default": {},
+          "title": "Refs",
+          "type": "object"
+        },
+        "source": {
+          "default": "",
+          "title": "Source",
+          "type": "string"
+        },
+        "title": {
+          "default": "",
+          "title": "Title",
+          "type": "string"
+        },
+        "url": {
+          "default": "",
+          "title": "Url",
+          "type": "string"
+        }
+      },
+      "required": [
+        "kind"
+      ],
+      "title": "NewTweetPayload",
+      "type": "object"
+    },
     "PluginActionInput": {
       "description": "Fallback input member for a plugin (non-built-in) action kind. Mirrors\nPluginActionWire on the write path; `config` is required (you can't author an\naction without one) but open (validated server-side by the kind's registry).",
       "properties": {
@@ -2440,7 +2561,8 @@
                   "hn_comment": "#/$defs/HackerNewsCommentSourceSpec",
                   "hn_feed": "#/$defs/HackerNewsFeedSourceSpec",
                   "reddit_subreddit": "#/$defs/RedditSubredditSourceSpec",
-                  "rss": "#/$defs/RssSourceSpec"
+                  "rss": "#/$defs/RssSourceSpec",
+                  "twitter_search": "#/$defs/TwitterSearchSourceSpec"
                 },
                 "propertyName": "kind"
               },
@@ -2456,6 +2578,9 @@
                 },
                 {
                   "$ref": "#/$defs/HackerNewsCommentSourceSpec"
+                },
+                {
+                  "$ref": "#/$defs/TwitterSearchSourceSpec"
                 }
               ]
             },
@@ -2582,7 +2707,8 @@
                   "hn_comment": "#/$defs/HackerNewsCommentSourceSpec",
                   "hn_feed": "#/$defs/HackerNewsFeedSourceSpec",
                   "reddit_subreddit": "#/$defs/RedditSubredditSourceSpec",
-                  "rss": "#/$defs/RssSourceSpec"
+                  "rss": "#/$defs/RssSourceSpec",
+                  "twitter_search": "#/$defs/TwitterSearchSourceSpec"
                 },
                 "propertyName": "kind"
               },
@@ -2598,6 +2724,9 @@
                 },
                 {
                   "$ref": "#/$defs/HackerNewsCommentSourceSpec"
+                },
+                {
+                  "$ref": "#/$defs/TwitterSearchSourceSpec"
                 }
               ]
             },
@@ -2647,6 +2776,48 @@
       "title": "TelemetryState",
       "type": "object"
     },
+    "TwitterSearchSourceSpec": {
+      "description": "Identity of one X (Twitter) search stream. Bound to TwitterSearchConnector.\n\n`query` is the search expression (keywords, quoted phrases, `from:`,\n`lang:`, `filter:` operators, whatever X's search syntax accepts); it is\nREQUIRED and NON-BLANK so a source always carries a server-side pre-filter\nbefore any per-item LLM cost (same discipline as hn_comment: a blank query\nwould be the unfiltered firehose). `mode` picks the result ordering twikit\nasks X for: `latest` (newest first, the listener's default) or `top`\n(ranked). `count` caps the per-cycle fetch. `lang` optionally narrows to\ntweets in one language (ISO 639-1, e.g. \"en\"); empty = no filter.",
+      "properties": {
+        "count": {
+          "default": 20,
+          "maximum": 100,
+          "minimum": 1,
+          "title": "Count",
+          "type": "integer"
+        },
+        "kind": {
+          "const": "twitter_search",
+          "default": "twitter_search",
+          "title": "Kind",
+          "type": "string"
+        },
+        "lang": {
+          "default": "",
+          "title": "Lang",
+          "type": "string"
+        },
+        "mode": {
+          "default": "latest",
+          "enum": [
+            "latest",
+            "top"
+          ],
+          "title": "Mode",
+          "type": "string"
+        },
+        "query": {
+          "minLength": 1,
+          "title": "Query",
+          "type": "string"
+        }
+      },
+      "required": [
+        "query"
+      ],
+      "title": "TwitterSearchSourceSpec",
+      "type": "object"
+    },
     "WatchActionBackfillState": {
       "description": "Lifecycle of one WatchActionBackfill job (a queued request to re-run an\naction over the previous step's passes).\n\nThe `process_due_backfills` cron claims a PENDING job (CAS -> RUNNING), does\nthe select/delete/enqueue, then marks it terminal:\n  - DONE   : the setup finished, and the enqueued runs are now the drain's job.\n  - FAILED : same dual meaning as a run's FAILED, transient-until-exhausted, and\n             (like a run) readable off `completed_at`: FAILED with `completed_at`\n             UNSET is retryable (the reaper cleared it so claim_due re-picks it);\n             FAILED with `completed_at` SET is terminal (attempts hit the cap). The\n             reaper resets a stale RUNNING to FAILED (retryable, `completed_at`\n             cleared), and a permanent setup defect (source action gone) fails with\n             attempts bumped to the cap AND `completed_at` stamped, so it isn't\n             re-claimed. Terminality is the attempts cap / `completed_at`, not the\n             state alone.\nA RUNNING job whose worker died is reaped to FAILED (retryable), so a crash\nmid-setup is retried, safe because the setup is idempotent and guarded by the\njob's `replace_deleted_at` delete-once marker.",
       "enum": [
@@ -3972,6 +4143,9 @@
     {
       "$ref": "#/$defs/HackerNewsCommentSourceSpec"
     },
+    {
+      "$ref": "#/$defs/TwitterSearchSourceSpec"
+    },
     {
       "$ref": "#/$defs/EngineStatus"
     },
diff --git a/packages/openmagpie-schema/src/openmagpie_schema/configs.py b/packages/openmagpie-schema/src/openmagpie_schema/configs.py
index 34258ef5..51e46d11 100644
--- a/packages/openmagpie-schema/src/openmagpie_schema/configs.py
+++ b/packages/openmagpie-schema/src/openmagpie_schema/configs.py
@@ -149,12 +149,55 @@ def display(self) -> str:
         return f'HN comments: "{self.query}"'
 
 
+class TwitterSearchSourceSpec(BaseModel):
+    """Identity of one X (Twitter) search stream. Bound to TwitterSearchConnector.
+
+    `query` is the search expression (keywords, quoted phrases, `from:`,
+    `lang:`, `filter:` operators, whatever X's search syntax accepts); it is
+    REQUIRED and NON-BLANK so a source always carries a server-side pre-filter
+    before any per-item LLM cost (same discipline as hn_comment: a blank query
+    would be the unfiltered firehose). `mode` picks the result ordering twikit
+    asks X for: `latest` (newest first, the listener's default) or `top`
+    (ranked). `count` caps the per-cycle fetch. `lang` optionally narrows to
+    tweets in one language (ISO 639-1, e.g. "en"); empty = no filter.
+    """
+
+    SOURCE_KIND: ClassVar[str] = "twitter_search"
+    URL_FIELDS: ClassVar[tuple[str, ...]] = ()  # no operator-supplied URL to SSRF-check
+
+    kind: Literal["twitter_search"] = "twitter_search"
+    query: str = Field(min_length=1)
+    mode: Literal["latest", "top"] = "latest"
+    count: int = Field(default=20, ge=1, le=100)
+    lang: str = ""
+
+    @field_validator("query")
+    @classmethod
+    def _query_not_blank(cls, v: str) -> str:
+        v = v.strip()
+        if not v:
+            raise ValueError("twitter_search requires a non-blank query (the firehose guard)")
+        return v
+
+    @field_validator("lang")
+    @classmethod
+    def _lang_normalize(cls, v: str) -> str:
+        return v.strip().lower()
+
+    def display(self) -> str:
+        return f'X search: "{self.query}"'
+
+
 # The built-ins as a discriminated union over `kind` (defined before the plugin
 # fallback so the built-in kind set can be derived from it below). A built-in kind
 # with a malformed spec fails its typed member here and is rejected by the fallback,
 # so it surfaces as a validation error rather than being absorbed as a raw blob.
 _BuiltinSourceSpec = Annotated[
-    RedditSubredditSourceSpec | RssSourceSpec | HackerNewsFeedSourceSpec | HackerNewsCommentSourceSpec,
+    RedditSubredditSourceSpec
+    | RssSourceSpec
+    | HackerNewsFeedSourceSpec
+    | HackerNewsCommentSourceSpec
+    | TwitterSearchSourceSpec,
     Field(discriminator="kind"),
 ]
 
diff --git a/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py b/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py
index 0ee8d27e..e8d7f574 100644
--- a/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py
+++ b/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py
@@ -79,6 +79,24 @@ class HackerNewsCommentPayload(FeedItemPayload):
     story_title: str = ""
 
 
+class NewTweetPayload(FeedItemPayload):
+    """`new_tweet`: one X (Twitter) tweet (TwitterSearchConnector, unofficial
+    twikit route). `content` is the tweet's full text (the engine's judgeable
+    body); `title` is empty (tweets have no headline). The listeningkit
+    SocialEvent shape is carried as typed fields: `handle` (the @screen_name,
+    also the within-kind source slug), `author` (display name), `lang`,
+    `metrics` (likes/retweets/replies/quotes/views), `refs`
+    (in_reply_to / quoted / retweet_of), `media` (list of {type,url,thumbnail})."""
+
+    kind: Literal["new_tweet"]  # required, so a non-twitter dump can't match here
+    author: str = ""
+    handle: str = ""
+    lang: str = ""
+    metrics: dict[str, int | None] = {}
+    refs: dict[str, str | None] = {}
+    media: list[dict[str, object]] = []
+
+
 # Tried left-to-right so a dump resolves to its concrete variant (matched on the
 # required `kind` literal) and only falls to the permissive base when no variant
 # claims it. Variants REQUIRE their `kind`, so an empty / kind-less dict can't
@@ -89,6 +107,11 @@ class HackerNewsCommentPayload(FeedItemPayload):
 # but a consumer keying on `isinstance(data, RssEntryPayload)` won't see the
 # malformed row (canonical fields like `title` still read off the base).
 FeedItemData = Annotated[
-    RssEntryPayload | NewRedditPostPayload | HackerNewsFeedPayload | HackerNewsCommentPayload | FeedItemPayload,
+    RssEntryPayload
+    | NewRedditPostPayload
+    | HackerNewsFeedPayload
+    | HackerNewsCommentPayload
+    | NewTweetPayload
+    | FeedItemPayload,
     Field(union_mode="left_to_right"),
 ]
diff --git a/tools/schema_sync/models.py b/tools/schema_sync/models.py
index 5be4e1fc..f0234281 100644
--- a/tools/schema_sync/models.py
+++ b/tools/schema_sync/models.py
@@ -19,6 +19,7 @@
     HackerNewsFeedSourceSpec,
     RedditSubredditSourceSpec,
     RssSourceSpec,
+    TwitterSearchSourceSpec,
 )
 from openmagpie_schema.engine import EngineListResponse, EngineStatus
 from openmagpie_schema.feed import (
@@ -107,6 +108,7 @@
     RssSourceSpec,
     HackerNewsFeedSourceSpec,
     HackerNewsCommentSourceSpec,
+    TwitterSearchSourceSpec,
     # Engine + telemetry status
     EngineStatus,
     EngineListResponse,
diff --git a/uv.lock b/uv.lock
index 1c5abdf6..4ba751ac 100644
--- a/uv.lock
+++ b/uv.lock
@@ -67,6 +67,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" },
 ]
 
+[[package]]
+name = "beautifulsoup4"
+version = "4.15.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "soupsieve" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/43/65/318323f98dbee45d42dff61d8f047181bc6f2268a9068cfad035a46be5af/beautifulsoup4-4.15.0.tar.gz", hash = "sha256:288e3ca7d54b06f2ac191970bc275c1939cb46d450b255bf6718b04aa37ab4f7", size = 632571, upload-time = "2026-06-07T16:44:20.453Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/88/c6/92fcd42f1ba33e1184263f25bfabf3d27c383410470f169e4b8163bf9c17/beautifulsoup4-4.15.0-py3-none-any.whl", hash = "sha256:d6f88de62e1d4e38ecb1077eb9724cd0eff29d2a08ca16a401e9b9e93f117cf9", size = 109924, upload-time = "2026-06-07T16:44:21.566Z" },
+]
+
 [[package]]
 name = "certifi"
 version = "2026.5.20"
@@ -372,6 +385,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/4e/eb/c96d64137e29ae17d83ad2552470bafe3a7a915e85434d9942077d7fd011/feedparser-6.0.12-py3-none-any.whl", hash = "sha256:6bbff10f5a52662c00a2e3f86a38928c37c48f77b3c511aedcd51de933549324", size = 81480, upload-time = "2025-09-10T13:33:58.022Z" },
 ]
 
+[[package]]
+name = "filetype"
+version = "1.2.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/bb/29/745f7d30d47fe0f251d3ad3dc2978a23141917661998763bebb6da007eb1/filetype-1.2.0.tar.gz", hash = "sha256:66b56cd6474bf41d8c54660347d37afcc3f7d1970648de365c102ef77548aadb", size = 998020, upload-time = "2022-11-02T17:34:04.141Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/18/79/1b8fa1bb3568781e84c9200f951c735f3f157429f44be0495da55894d620/filetype-1.2.0-py2.py3-none-any.whl", hash = "sha256:7ce71b6880181241cf7ac8697a2f1eb6a8bd9b429f7ad6d27b8db9ba5f1c2d25", size = 19970, upload-time = "2022-11-02T17:34:01.425Z" },
+]
+
 [[package]]
 name = "gunicorn"
 version = "26.0.0"
@@ -437,6 +459,11 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
 ]
 
+[package.optional-dependencies]
+socks = [
+    { name = "socksio" },
+]
+
 [[package]]
 name = "idna"
 version = "3.18"
@@ -500,6 +527,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/0e/a4/cf8d779feb133a27a2e3bc833bccb9e13aa332cdf820497ebf72c10ce8c3/jiter-0.15.0-cp314-cp314t-win_arm64.whl", hash = "sha256:66b1880df2d01e206e8339769d1c7c1753bcb653efd6289e203f6f24ebada0c0", size = 191244, upload-time = "2026-05-19T10:09:12.74Z" },
 ]
 
+[[package]]
+name = "js2py-3-13"
+version = "0.74.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pyjsparser" },
+    { name = "six" },
+    { name = "tzlocal" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/e2/9e/17ed2ceebff1539a454b66d3f056001ab37f679e28a336e1ba88407940fe/js2py_3_13-0.74.1.tar.gz", hash = "sha256:91e214f717312f9d510eaf36fcc5325b0b15a22a49831fe2b434bca4a33c1f77", size = 570397, upload-time = "2025-02-07T13:01:33.395Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/47/5f/4bdab35d30055613c58f681f03d3a76e06c874485aced646fac763a1d552/Js2Py_3.13-0.74.1-py3-none-any.whl", hash = "sha256:5c60a80a43197775986c27f33becaf9ebf3731e8e79030c925f44025ed6f0e8b", size = 611795, upload-time = "2025-02-07T13:01:31.087Z" },
+]
+
 [[package]]
 name = "justext"
 version = "3.0.2"
@@ -604,6 +645,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/6a/bd/6e2b76a6c5dee10397db9c929f0c5066766ec1036046f0335b7ca7ca08b8/lxml_html_clean-0.4.5-py3-none-any.whl", hash = "sha256:c76fcadd1e5bfb9b8bafc2200d51e4e78eb0dad67f56881c21dfb6484c7e7746", size = 14573, upload-time = "2026-05-20T12:17:52.215Z" },
 ]
 
+[[package]]
+name = "m3u8"
+version = "6.0.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/9b/a5/73697aaa99bb32b610adc1f11d46a0c0c370351292e9b271755084a145e6/m3u8-6.0.0.tar.gz", hash = "sha256:7ade990a1667d7a653bcaf9413b16c3eb5cd618982ff46aaff57fe6d9fa9c0fd", size = 42720, upload-time = "2024-08-07T11:20:06.606Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/f8/31/50f3c38b38ff28635ff9c4a4afefddccc5f1b57457b539bdbdf75ce18669/m3u8-6.0.0-py3-none-any.whl", hash = "sha256:566d0748739c552dad10f8c87150078de6a0ec25071fa48e6968e96fc6dcba5d", size = 24133, upload-time = "2024-08-07T11:20:03.96Z" },
+]
+
 [[package]]
 name = "markdown-it-py"
 version = "4.2.0"
@@ -706,6 +756,7 @@ dependencies = [
     { name = "python-dotenv" },
     { name = "pyyaml" },
     { name = "trafilatura" },
+    { name = "twikit" },
     { name = "ulid" },
 ]
 
@@ -739,6 +790,7 @@ requires-dist = [
     { name = "python-dotenv", specifier = ">=1.1" },
     { name = "pyyaml", specifier = ">=6.0" },
     { name = "trafilatura", specifier = ">=1.7" },
+    { name = "twikit", git = "https://github.com/unclecode/twikit.git" },
     { name = "ulid", specifier = ">=1.1" },
 ]
 
@@ -939,6 +991,21 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" },
 ]
 
+[[package]]
+name = "pyjsparser"
+version = "2.7.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/48/ef/c72abcfa2c6accd03e7c89c400790fc3d908c5804d50a7c4e9ceabd74d23/pyjsparser-2.7.1.tar.gz", hash = "sha256:be60da6b778cc5a5296a69d8e7d614f1f870faf94e1b1b6ac591f2ad5d729579", size = 24196, upload-time = "2019-04-21T21:56:17.708Z" }
+
+[[package]]
+name = "pyotp"
+version = "2.10.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/2d/c6/c5d96a86fd0bf6fa1bbb5c5c341ff3208638b692727a683c8289068d9a11/pyotp-2.10.0.tar.gz", hash = "sha256:d01e9703443616b03c57c700b5cbffd56a1f929c1b0f8f03131bc78c1fca9d3f", size = 18625, upload-time = "2026-06-14T03:48:49.221Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/65/33/7b83bde70eddaaaaef487751a9c3a5cc0c0be54620ded0e120ebdc401ff9/pyotp-2.10.0-py3-none-any.whl", hash = "sha256:1df2f6a1bcc3bb0716172a5215ddc2f8c7c7fd26a13df9927d52e1746934836c", size = 13768, upload-time = "2026-06-14T03:48:47.831Z" },
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -1163,6 +1230,24 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235, upload-time = "2024-02-25T23:20:01.196Z" },
 ]
 
+[[package]]
+name = "socksio"
+version = "1.0.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/5c/48a7d9495be3d1c651198fd99dbb6ce190e2274d0f28b9051307bdec6b85/socksio-1.0.0.tar.gz", hash = "sha256:f88beb3da5b5c38b9890469de67d0cb0f9d494b78b106ca1845f96c10b91c4ac", size = 19055, upload-time = "2020-04-17T15:50:34.664Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/37/c3/6eeb6034408dac0fa653d126c9204ade96b819c936e136c5e8a6897eee9c/socksio-1.0.0-py3-none-any.whl", hash = "sha256:95dc1f15f9b34e8d7b16f06d74b8ccf48f609af32ab33c608d08761c5dcbb1f3", size = 12763, upload-time = "2020-04-17T15:50:31.878Z" },
+]
+
+[[package]]
+name = "soupsieve"
+version = "2.9.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/69/99/a6ca3beb3ccacb41fb3321d8a60e5566f9e6467601ef8eba6a17e1b89778/soupsieve-2.9.2.tar.gz", hash = "sha256:4a55d8cf158a9c2e587fa4922f1bbb91d68ac829e2d6f25403a85747c71daf74", size = 122445, upload-time = "2026-08-07T00:57:24.801Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/eb/dc/ad025c1ee131eba60c69f4dd5779b18fcf1e6b21a343e2162a84d5d133c7/soupsieve-2.9.2-py3-none-any.whl", hash = "sha256:8089a26fd974ca7a1f30276d3d8492ab266ab15af581642dfe8aa162e0c1c823", size = 37370, upload-time = "2026-08-07T00:57:23.524Z" },
+]
+
 [[package]]
 name = "sqlparse"
 version = "0.5.5"
@@ -1211,6 +1296,21 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/0e/78/4ad99d79aee2784f49f20fd0a29058ce4c032fe4439047924c43521cd211/trafilatura-2.1.0-py3-none-any.whl", hash = "sha256:0eded5207a806445ddebbe36eae30b9035fe6a2f233c36f6fe82663fca8b9d30", size = 134600, upload-time = "2026-06-07T17:43:28.404Z" },
 ]
 
+[[package]]
+name = "twikit"
+version = "2.3.3"
+source = { git = "https://github.com/unclecode/twikit.git#6a73ab97f4de09f79139f6308c9fb80029a9f5f7" }
+dependencies = [
+    { name = "beautifulsoup4" },
+    { name = "filetype" },
+    { name = "httpx", extra = ["socks"] },
+    { name = "js2py-3-13" },
+    { name = "lxml" },
+    { name = "m3u8" },
+    { name = "pyotp" },
+    { name = "webvtt-py" },
+]
+
 [[package]]
 name = "ty"
 version = "0.0.43"
@@ -1388,3 +1488,12 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/25/91/80908e835e100527a9267147b08c0eee1fa6ab0ffec15edc04d1d44885f7/watchfiles-1.2.0-cp315-cp315-musllinux_1_1_aarch64.whl", hash = "sha256:b718bf356bbc15e559bd8ef41782b573b8ae0e3f177ab244b440568d7ea02cfb", size = 630638, upload-time = "2026-05-18T04:30:49.89Z" },
     { url = "https://files.pythonhosted.org/packages/46/4b/95ab2f256bb4af3cb2eb23b9317bda984ee6e0f11733a5c004a6c95b06e3/watchfiles-1.2.0-cp315-cp315-musllinux_1_1_x86_64.whl", hash = "sha256:922c0e019fe68b3ae392965a766b02a71ba1168c932cebc3733cd52c5fe5b377", size = 657684, upload-time = "2026-05-18T04:31:32.027Z" },
 ]
+
+[[package]]
+name = "webvtt-py"
+version = "0.5.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/5e/f6/7c9c964681fb148e0293e6860108d378e09ccab2218f9063fd3eb87f840a/webvtt-py-0.5.1.tar.gz", hash = "sha256:2040dd325277ddadc1e0c6cc66cbc4a1d9b6b49b24c57a0c3364374c3e8a3dc1", size = 55128, upload-time = "2024-05-30T13:40:17.189Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/f3/ed/aad7e0f5a462d679f7b4d2e0d8502c3096740c883b5bbed5103146480937/webvtt_py-0.5.1-py3-none-any.whl", hash = "sha256:9d517d286cfe7fc7825e9d4e2079647ce32f5678eb58e39ef544ffbb932610b7", size = 19802, upload-time = "2024-05-30T13:40:14.661Z" },
+]

From e611a6cc8d9c72329db1986fc95a7f7fa2d83d9d Mon Sep 17 00:00:00 2001
From: matthewdonsemail-lab <matthewdonsemail@gmail.com>
Date: Sat, 8 Aug 2026 19:04:13 +0700
Subject: [PATCH 2/6] fix(sources/twitter): per-call twikit client + retryable
 empty-404 mapping

Live X polling through the connector exposed two issues (verified against
live cookies from listeningkit/backend/credentials/personal.json, no
proxy, no vendor API):

1. Event loop is closed on 2nd/3rd source in a multi-source feed poll.
   TwikitClient built one twikit Client at import time; twikit's Client
   creates an httpx.AsyncClient bound to the *currently running* event
   loop, and TwikitClient.search() drives it with a fresh asyncio.run
   loop per call. The shared client died with the first loop, so later
   sources in the same poll failed. Fix: construct the twikit Client
   inside _search_async (per call, inside the running loop); cookies are
   still resolved once in __init__.

2. X SearchTimeline intermittently 404s with an EMPTY body (reproduced:
   same query succeeds on immediate retry, session-independent). That is
   a transient upstream flake, not a deleted tweet/user; the old
   not_found/non-retryable mapping made ops treat a healthy source as
   dead. Fix: empty-body NotFound maps to search_timeline_unavailable
   (retryable=True, backoff); message-bearing NotFound stays not_found.

- client.py: per-call twikit Client construction + docstring.
- errors.py: empty-404 special case in map_twikit_error.
- tests_twitter.py: regression test for the retryable empty-404 mapping.

Validation: ruff check + format (416 files), ty, makemigrations --check,
schema.json --check, full Django suite 587 tests OK, live poll of the
x-buying-signals feed (4 twitter_search sources) green end-to-end.

Signed-off-by: matthewdonsemail-lab <matthewdonsemail@gmail.com>
Co-authored-by: Matthew Don <matthewdonsemail@gmail.com>
---
 .../core/sources/connectors/twitter/client.py | 33 ++++++++++++-------
 .../core/sources/connectors/twitter/errors.py | 17 ++++++++++
 apps/core/sources/tests_twitter.py            | 19 +++++++++++
 3 files changed, 57 insertions(+), 12 deletions(-)

diff --git a/apps/core/sources/connectors/twitter/client.py b/apps/core/sources/connectors/twitter/client.py
index 9905d01e..48ea7e81 100644
--- a/apps/core/sources/connectors/twitter/client.py
+++ b/apps/core/sources/connectors/twitter/client.py
@@ -127,7 +127,18 @@ def __init__(self, err: ListenerError) -> None:
 
 
 class TwikitClient:
-    """Thin, proxy-bound wrapper around the twikit async client."""
+    """Thin, proxy-bound wrapper around the twikit async client.
+
+    One twikit ``Client`` per search call. twikit's ``Client.__init__``
+    creates an ``httpx.AsyncClient`` bound to the *currently running*
+    event loop, and ``search()`` drives twikit with a fresh
+    ``asyncio.run`` loop per call — so the twikit client must be
+    constructed inside that loop. A shared instance created at import
+    time dies with the first loop and later searches fail with
+    "Event loop is closed" (observed on the second/third source in a
+    multi-source feed poll). Cookies are resolved once here (cheap env /
+    file reads); only the twikit client is per-call.
+    """
 
     def __init__(
         self,
@@ -140,6 +151,8 @@ def __init__(
         cookies_file: str | None = None,
         credentials_dir: str | None = None,
     ) -> None:
+        self._language = language
+        self._user_agent = user_agent
         self.proxy = proxy
         if cookies is not None:
             self._cookies = dict(cookies)
@@ -149,23 +162,19 @@ def __init__(
                 cookies_file=cookies_file,
                 credentials_dir=credentials_dir,
             )
-        self._client = Client(language=language, proxy=self.proxy, user_agent=user_agent)
-        self._ready = False
 
-    def _ensure_session(self) -> None:
-        if self._ready:
-            return
+    async def _search_async(self, query: str, mode: TwikitProduct, count: int):
+        # Build the twikit client here, inside the event loop search()
+        # runs: twikit's Client binds its httpx.AsyncClient to this loop
+        # at construction (see class docstring).
+        client = Client(language=self._language, proxy=self.proxy, user_agent=self._user_agent)
         if self._cookies:
-            self._client.set_cookies(dict(self._cookies), clear_cookies=True)
+            client.set_cookies(dict(self._cookies), clear_cookies=True)
             log.info("session: loaded %d cookie(s)", len(self._cookies))
         else:
             log.warning("session: no cookies configured; guest mode only")
-        self._ready = True
-
-    async def _search_async(self, query: str, mode: TwikitProduct, count: int):
-        self._ensure_session()
         try:
-            return await self._client.search_tweet(query, mode, count=count)
+            return await client.search_tweet(query, mode, count=count)
         except TwitterException as exc:
             raise ListenerErrorWrapper(map_twikit_error(exc, {"query": query, "mode": mode})) from exc
         except Exception as exc:  # bootstrap failures (degraded shell) are not TwitterException
diff --git a/apps/core/sources/connectors/twitter/errors.py b/apps/core/sources/connectors/twitter/errors.py
index f3e7578e..0fb45b08 100644
--- a/apps/core/sources/connectors/twitter/errors.py
+++ b/apps/core/sources/connectors/twitter/errors.py
@@ -94,6 +94,23 @@ def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = Non
     reset = getattr(exc, "rate_limit_reset", None)
     headers = getattr(exc, "headers", None)
 
+    # X's SearchTimeline intermittently 404s with an EMPTY body (observed
+    # repeatedly on live polls: same query succeeds on retry seconds later,
+    # independent of session/cookies/query). That is a transient upstream
+    # flake on the search endpoint, NOT a deleted tweet/user — a genuine
+    # not_found carries a message. Retryable so the ops layer backs off
+    # instead of treating the source as dead.
+    if isinstance(exc, NotFound) and not str(exc).strip():
+        return ListenerError(
+            code="search_timeline_unavailable",
+            message="X SearchTimeline returned an empty 404 (transient upstream flake)",
+            retryable=True,
+            action="retry with backoff; watermark stays put so the next cycle re-reads",
+            context=context or {},
+            headers=headers,
+            rate_limit_reset=reset,
+        )
+
     retryable_actions: dict[str, tuple[bool, str]] = {
         "bad_request": (False, "fix query / payload; do not retry as-is"),
         "unauthorized": (False, "refresh session (guest token / cookies) and retry once"),
diff --git a/apps/core/sources/tests_twitter.py b/apps/core/sources/tests_twitter.py
index 217be08b..aa0bfee0 100644
--- a/apps/core/sources/tests_twitter.py
+++ b/apps/core/sources/tests_twitter.py
@@ -111,6 +111,25 @@ def test_error_maps_to_connector_parse_error(self):
             list(conn.poll(spec, since=None))
         self.assertIn("rate_limited", str(ctx.exception))
 
+    def test_empty_404_maps_to_retryable_connector_error(self):
+        """X SearchTimeline empty-404 is a transient flake, not a dead tweet."""
+        from sources.connectors.twitter.errors import ListenerError
+
+        spec = TwitterSearchSourceSpec(kind="twitter_search", query="x")
+        err = ListenerError(
+            code="search_timeline_unavailable",
+            message="X SearchTimeline returned an empty 404 (transient upstream flake)",
+            retryable=True,
+            action="retry with backoff",
+        )
+        client = mock.Mock()
+        client.search.side_effect = ListenerErrorWrapper(err)
+        conn = TwitterSearchConnector()
+        conn._client = client
+        with self.assertRaises(ConnectorParseError) as ctx:
+            list(conn.poll(spec, since=None))
+        self.assertIn("search_timeline_unavailable", str(ctx.exception))
+
     def test_mode_top_maps_to_twikit_top(self):
         spec = TwitterSearchSourceSpec(kind="twitter_search", query="x", mode="top")
         conn, client = self._connector([_FakeTweet("1")])

From 785cd4daf4d7b6a47d1e2aba46a61b90b449e7c1 Mon Sep 17 00:00:00 2001
From: matthewdonsemail-lab <matthewdonsemail@gmail.com>
Date: Sat, 8 Aug 2026 19:27:09 +0700
Subject: [PATCH 3/6] docs(readme): add X/Twitter to the platform diagram,
 upcoming platforms, and a what-we've-done outline

Brings the README in line with the twitter_search connector work shipped
on this branch:

- Diagram + 'Where it listens' now show X/Twitter as a shipped source,
  and Facebook, TikTok, and Instagram as 'soon to be added'.
- 'What's shipped today' lists the twitter_search source kind.
- New 'What we've done' section outlines the twikit connector, live-cookie
  auth, the live-poll reliability fixes, and the verified end-to-end
  feed -> watch -> webhook delivery (44/44 HTTP 200, payload matched
  against the Twenty socialEvent intake contract).
- Roadmap's connector list updated to match (X shipped; FB/TikTok/IG next).

Co-authored-by: Matthew Don <matthewdonsemail@gmail.com>
Signed-off-by: Matthew Don <matthewdonsemail@gmail.com>
---
 README.md | 28 ++++++++++++++++++++++++----
 1 file changed, 24 insertions(+), 4 deletions(-)

diff --git a/README.md b/README.md
index 3fd5b1b1..dbcf98dc 100644
--- a/README.md
+++ b/README.md
@@ -28,7 +28,7 @@
 
 ## What it does
 
-You scan Reddit, Hacker News, and a few RSS feeds looking for someone hitting a problem your product solves or asking a question you can answer well. Getting there while the conversation is happening is how you build a brand and a community around what you know. OpenMagpie watches the threads for you so you spend your time on engagement instead of searching.
+You scan X/Twitter, Reddit, Hacker News, and a few RSS feeds looking for someone hitting a problem your product solves or asking a question you can answer well. Getting there while the conversation is happening is how you build a brand and a community around what you know. OpenMagpie watches the threads for you so you spend your time on engagement instead of searching.
 
 You curate sources into a feed, write a natural-language description of what's relevant (for example, "someone frustrated with manual social monitoring and asking for alternatives"), and a local LLM run via any OpenAI-compatible runner (e.g. Ollama, vLLM, LM Studio) scores each new post against it. Matches go to a webhook or your logs (more integrations coming); everything else is dropped. You read the hits instead of the firehose.
 
@@ -36,8 +36,9 @@ You curate sources into a feed, write a natural-language description of what's r
 
 OpenMagpie listens wherever communities are having those conversations.
 
-- **Public discussion (today):** Reddit, Hacker News, and any RSS or Atom feed (news, blogs, Substack publications, and forums that publish feeds).
+- **Public discussion (today):** X/Twitter, Reddit, Hacker News, and any RSS or Atom feed (news, blogs, Substack publications, and forums that publish feeds).
 - **Communities you're in (roadmap):** Slack workspaces and LinkedIn you already belong to, so you catch relevant threads in the groups where you participate, no admin or app install required.
+- **Public discussion (soon to be added):** Facebook, TikTok, and Instagram.
 
 ## Quickstart
 
@@ -142,12 +143,16 @@ A `Feed` is a reusable, curated stream (a set of sources plus an item log). A `W
 ```mermaid
 graph TD
     subgraph Sources
+        TWITTER[X / Twitter]
         REDDIT[Reddit]
         RSS[RSS / Atom feeds]
         HN[Hacker News]
         SLACK[Slack]
         LINKEDIN[LinkedIn]
         GITHUB[GitHub]
+        FACEBOOK[Facebook]
+        TIKTOK[TikTok]
+        INSTAGRAM[Instagram]
     end
 
     subgraph OpenMagpie
@@ -164,12 +169,16 @@ graph TD
         FUTURE["email / Slack (planned)"]
     end
 
+    TWITTER --> FEED
     REDDIT --> FEED
     RSS --> FEED
     HN --> FEED
     SLACK -. planned .-> FEED
     LINKEDIN -. planned .-> FEED
     GITHUB -. planned .-> FEED
+    FACEBOOK -. soon to be added .-> FEED
+    TIKTOK -. soon to be added .-> FEED
+    INSTAGRAM -. soon to be added .-> FEED
 
     FEED -- "new items" --> WATCH
     WATCH -- "action chain" --> FILTER
@@ -222,16 +231,27 @@ Social listening is a crowded market (Brand24, Mention, Octolens, Syften, and to
 
 | Layer | Shipped |
 |---|---|
-| Connectors | Reddit (`reddit_subreddit`), Hacker News (`hn_feed`, `hn_comment`), RSS/Atom (`rss`) |
+| Connectors | X/Twitter (`twitter_search`), Reddit (`reddit_subreddit`), Hacker News (`hn_feed`, `hn_comment`), RSS/Atom (`rss`) |
 | Engines | Any OpenAI-compatible `/v1` API: Ollama, vLLM, llama.cpp, LM Studio, OpenAI, ... |
 | Action kinds | `semantic_filter` (LLM-judged), `webhook`, `log` |
 | Delivery modes | instant, digest |
 | 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`).
+
+Next up on the roadmap: **Facebook, TikTok, and Instagram connectors** (soon to be added), then Slack, LinkedIn, GitHub, Bluesky, and Mastodon.
+
 ## Roadmap
 
-- **More connectors**: Slack, LinkedIn, GitHub, Bluesky, Mastodon, and X.
+- **More connectors**: Facebook, TikTok, and Instagram (soon to be added), then Slack, LinkedIn, GitHub, Bluesky, and Mastodon.
 - **More engines**: Anthropic, OpenAI, and a keyword engine behind the same `Engine` Protocol.
 - **Learns from feedback**: thumbs up/down on past matches become few-shot examples for the next pass.
 - **Run-history in the payload**: the upstream filter score and chain provenance as an opt-in webhook field.

From 928bd82a1b4267ea9894a5b5e94e0eeb6889eff6 Mon Sep 17 00:00:00 2001
From: matthewdonsemail-lab <matthewdonsemail@gmail.com>
Date: Sat, 8 Aug 2026 19:28:30 +0700
Subject: [PATCH 4/6] docs(readme): quote soon-to-be-added edge labels in the
 mermaid diagram

Mermaid treats unquoted edge labels with spaces as separate tokens; quote
them to keep the diagram parseable on GitHub's renderer.

Co-authored-by: Matthew Don <matthewdonsemail@gmail.com>
Signed-off-by: Matthew Don <matthewdonsemail@gmail.com>
---
 README.md | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index dbcf98dc..b1e0de07 100644
--- a/README.md
+++ b/README.md
@@ -176,9 +176,9 @@ graph TD
     SLACK -. planned .-> FEED
     LINKEDIN -. planned .-> FEED
     GITHUB -. planned .-> FEED
-    FACEBOOK -. soon to be added .-> FEED
-    TIKTOK -. soon to be added .-> FEED
-    INSTAGRAM -. soon to be added .-> FEED
+    FACEBOOK -. "soon to be added" .-> FEED
+    TIKTOK -. "soon to be added" .-> FEED
+    INSTAGRAM -. "soon to be added" .-> FEED
 
     FEED -- "new items" --> WATCH
     WATCH -- "action chain" --> FILTER

From b8a8a171e5d1439c611f0e19759af13522854ff4 Mon Sep 17 00:00:00 2001
From: matthewdonsemail-lab <matthewdonsemail@gmail.com>
Date: Sat, 8 Aug 2026 20:08:07 +0700
Subject: [PATCH 5/6] feat(watches/webhook): carry source pattern_id in webhook
 payload

Adds optional pattern_id to the webhook body's per-item source object
(WebhookSource), sourced from FeedItem.source_meta (operator-supplied Source.meta
tags copied onto each item at record time). Receivers can now attribute yield by
listening pattern - e.g. the Twenty socialEvent intake's patternId field -
without deriving it by convention.

- schema: WebhookSource.pattern_id (str | None, default None) - the legacy
  contract (label+kind only) still validates; untagged sources emit null
- ActionItem gains source_meta (defaulted) so the action layer can read the tag
- run_inputs populates source_meta from the FeedItem column
- webhook _build_payload maps source_meta.pattern_id -> source.pattern_id
- tests: tagged -> pattern_id present; untagged -> null; legacy payload valid

Co-authored-by: Matthew Don <matthewdonsemail@gmail.com>
Signed-off-by: Matthew Don <matthewdonsemail@gmail.com>
---
 apps/core/watches/actions/protocol.py         |  1 +
 apps/core/watches/actions/webhook.py          |  4 ++-
 apps/core/watches/operations/run_inputs.py    |  1 +
 apps/core/watches/tests_webhook.py            | 33 ++++++++++++++++++-
 .../watch_actions/webhook.py                  |  9 +++--
 5 files changed, 44 insertions(+), 4 deletions(-)

diff --git a/apps/core/watches/actions/protocol.py b/apps/core/watches/actions/protocol.py
index a42ee8c4..a8fe9c44 100644
--- a/apps/core/watches/actions/protocol.py
+++ b/apps/core/watches/actions/protocol.py
@@ -38,6 +38,7 @@ class ActionItem:
     key: str
     source_label: str
     source_kind: str
+    source_meta: dict = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
diff --git a/apps/core/watches/actions/webhook.py b/apps/core/watches/actions/webhook.py
index 98d33d6e..fe50cf92 100644
--- a/apps/core/watches/actions/webhook.py
+++ b/apps/core/watches/actions/webhook.py
@@ -179,7 +179,9 @@ def _build_payload(
         items=[
             WebhookItem(
                 key=it.key,
-                source=WebhookSource(label=it.source_label, kind=it.source_kind),
+                source=WebhookSource(
+                    label=it.source_label, kind=it.source_kind, pattern_id=it.source_meta.get("pattern_id")
+                ),
                 item=_filtered(it.data, config.include_fields),
             )
             for it in items
diff --git a/apps/core/watches/operations/run_inputs.py b/apps/core/watches/operations/run_inputs.py
index 6261185a..4f9563cd 100644
--- a/apps/core/watches/operations/run_inputs.py
+++ b/apps/core/watches/operations/run_inputs.py
@@ -37,6 +37,7 @@ def build_run_inputs(
             key=f"{item.source_kind}:{item.external_id}",
             source_label=item.source_label,
             source_kind=item.source_kind,
+            source_meta=item.source_meta,
         )
         for _run, item in pairs
     ]
diff --git a/apps/core/watches/tests_webhook.py b/apps/core/watches/tests_webhook.py
index f5592be3..4e03ea24 100644
--- a/apps/core/watches/tests_webhook.py
+++ b/apps/core/watches/tests_webhook.py
@@ -102,9 +102,40 @@ def test_payload_is_self_describing(self) -> None:
         self.assertIsNotNone(body["window"])
         (sent,) = body["items"]
         self.assertEqual(sent["key"], "reddit:abc")
-        self.assertEqual(sent["source"], {"label": "r/ClaudeAI", "kind": "reddit_subreddit"})
+        self.assertEqual(sent["source"], {"label": "r/ClaudeAI", "kind": "reddit_subreddit", "pattern_id": None})
         self.assertEqual(sent["item"], {"title": "T"})  # url dropped by include_fields
 
+    def test_pattern_id_flows_from_source_meta(self) -> None:
+        # The contract extension: operator-supplied pattern_id on the source
+        # (FeedItem.source_meta, copied from Source.meta at record time) rides
+        # through to the wire body's per-item source so receivers can attribute
+        # yield by listening pattern. Absent meta -> pattern_id None, never a
+        # missing key (the shape stays self-describing).
+        action = WatchAction(id=ulid.ulid(), kind="webhook", config={"url": "https://h.example.com/hook"})
+        tagged = ActionItem(
+            data={"source": "twitter_search", "external_id": "t1"},
+            key="twitter_search:t1",
+            source_label="automation.manually",
+            source_kind="twitter_search",
+            source_meta={"pattern_id": "automation.manually", "lane": "AUTOMATION_BUILD"},
+        )
+        untagged = ActionItem(
+            data={"source": "twitter_search", "external_id": "t2"},
+            key="twitter_search:t2",
+            source_label="general-listener",
+            source_kind="twitter_search",
+        )
+        context = ActionContext(watch_id="w", watch_name="n", delivery=DeliveryCadence.INSTANT)
+        body = self._capture_request(action, tagged, context)["json"]
+        (tagged_sent,) = body["items"]
+        self.assertEqual(tagged_sent["source"]["pattern_id"], "automation.manually")
+        body = self._capture_request(action, untagged, context)["json"]
+        (untagged_sent,) = body["items"]
+        self.assertEqual(untagged_sent["source"]["pattern_id"], None)
+        self.assertEqual(
+            untagged_sent["source"], {"label": "general-listener", "kind": "twitter_search", "pattern_id": None}
+        )
+
     def test_method_is_dispatched(self) -> None:
         # A configured PUT is the verb actually sent (not hard-coded POST).
         action = WatchAction(
diff --git a/packages/openmagpie-schema/src/openmagpie_schema/watch_actions/webhook.py b/packages/openmagpie-schema/src/openmagpie_schema/watch_actions/webhook.py
index 482c82a5..e62116da 100644
--- a/packages/openmagpie-schema/src/openmagpie_schema/watch_actions/webhook.py
+++ b/packages/openmagpie-schema/src/openmagpie_schema/watch_actions/webhook.py
@@ -135,11 +135,16 @@ class WebhookWindow(BaseModel):
 
 
 class WebhookSource(BaseModel):
-    """Which feed source an item came from: the source's display `label` and
-    its connector `kind`."""
+    """Which feed source an item came from: the source's display `label`, its
+    connector `kind`, and the optional operator-supplied `pattern_id` tag
+    (yield attribution: which listening pattern produced this item).
+    `pattern_id` is None when the source carries no pattern tag (e.g. a
+    non-listening source) ; receivers map it to their pattern-attribution
+    field when present."""
 
     label: str
     kind: str
+    pattern_id: str | None = None
 
 
 class WebhookItem(BaseModel):

From 654e1769b761f4766319b63636529536c0b7a944 Mon Sep 17 00:00:00 2001
From: Prime Agent
 <1a1411c80158ee173e4906adc924e5bab025c48ebc1a0a780d2033451e8b7fc1@inferencesaver.communities.buzz.xyz>
Date: Sun, 9 Aug 2026 11:35:15 +0700
Subject: [PATCH 6/6] fix(sources/twitter): detect the empty-404 flake by its
 rendered body

twikit 2.3.3 renders every HTTP error as `status: <code>, message:
"<body>"` (client/client.py and guest/client.py), so str(exc) is never
empty, not even for an empty body. The e611a6c guard
`not str(exc).strip()` therefore never fired: live polls showed the
transient X SearchTimeline empty-404 mapping to not_found
(non-retryable), which made ops treat a healthy source as dead.

Match the empty BODY instead: an empty body renders the trailing payload
`message: ""`; a real 404 carries response text.

- errors.py: empty-404 detection checks the rendered body suffix.
- tests_twitter.py: connector-level empty-404 test now routes through a
  real twikit NotFound + map_twikit_error; added map-level tests pinning
  empty-body 404 -> search_timeline_unavailable (retryable) and
  message-bearing 404 -> not_found (non-retryable).

Validation: ruff check + format, ty, whitespace check, twitter tests
(12) and full core suite (590) OK.

Co-authored-by: Matthew Don <matthewdonsemail@gmail.com>
Signed-off-by: Matthew Don <matthewdonsemail@gmail.com>
---
 .../core/sources/connectors/twitter/errors.py | 11 +++++--
 apps/core/sources/tests_twitter.py            | 31 +++++++++++++------
 2 files changed, 31 insertions(+), 11 deletions(-)

diff --git a/apps/core/sources/connectors/twitter/errors.py b/apps/core/sources/connectors/twitter/errors.py
index 0fb45b08..7f051245 100644
--- a/apps/core/sources/connectors/twitter/errors.py
+++ b/apps/core/sources/connectors/twitter/errors.py
@@ -97,10 +97,17 @@ def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = Non
     # X's SearchTimeline intermittently 404s with an EMPTY body (observed
     # repeatedly on live polls: same query succeeds on retry seconds later,
     # independent of session/cookies/query). That is a transient upstream
-    # flake on the search endpoint, NOT a deleted tweet/user — a genuine
+    # flake on the search endpoint, NOT a deleted tweet/user: a genuine
     # not_found carries a message. Retryable so the ops layer backs off
     # instead of treating the source as dead.
-    if isinstance(exc, NotFound) and not str(exc).strip():
+    #
+    # twikit 2.3.3 renders every HTTP error as `status: <code>, message:
+    # "<body>"` (client/client.py and guest/client.py), so str(exc) is NEVER
+    # empty, not even for an empty body. Detect the empty BODY via its
+    # rendering instead: an empty body produces the trailing payload
+    # `message: ""`; any other rendering carried response text, so the 404
+    # is real.
+    if isinstance(exc, NotFound) and str(exc).rstrip().endswith('message: ""'):
         return ListenerError(
             code="search_timeline_unavailable",
             message="X SearchTimeline returned an empty 404 (transient upstream flake)",
diff --git a/apps/core/sources/tests_twitter.py b/apps/core/sources/tests_twitter.py
index aa0bfee0..be6e5b4b 100644
--- a/apps/core/sources/tests_twitter.py
+++ b/apps/core/sources/tests_twitter.py
@@ -12,12 +12,13 @@
 
 from django.test import SimpleTestCase
 from pydantic import ValidationError
+from twikit.errors import NotFound
 
 from openmagpie_schema.configs import TwitterSearchSourceSpec
 from sources.connectors.base import ConnectorParseError
 from sources.connectors.twitter.client import ListenerErrorWrapper
 from sources.connectors.twitter.connector import TwitterSearchConnector
-from sources.connectors.twitter.errors import ListenerError
+from sources.connectors.twitter.errors import ListenerError, map_twikit_error
 from sources.connectors.twitter.payloads import NewTweetPayload
 
 
@@ -113,15 +114,10 @@ def test_error_maps_to_connector_parse_error(self):
 
     def test_empty_404_maps_to_retryable_connector_error(self):
         """X SearchTimeline empty-404 is a transient flake, not a dead tweet."""
-        from sources.connectors.twitter.errors import ListenerError
-
         spec = TwitterSearchSourceSpec(kind="twitter_search", query="x")
-        err = ListenerError(
-            code="search_timeline_unavailable",
-            message="X SearchTimeline returned an empty 404 (transient upstream flake)",
-            retryable=True,
-            action="retry with backoff",
-        )
+        # twikit renders an empty-body 404 as 'status: 404, message: ""' (see
+        # client/client.py: message = f'status: {code}, message: "{text}"').
+        err = map_twikit_error(NotFound('status: 404, message: ""'))
         client = mock.Mock()
         client.search.side_effect = ListenerErrorWrapper(err)
         conn = TwitterSearchConnector()
@@ -137,6 +133,23 @@ def test_mode_top_maps_to_twikit_top(self):
         client.search.assert_called_once_with("x", "Top", 20)
 
 
+class MapTwikitErrorTests(SimpleTestCase):
+    """map_twikit_error: twikit's NotFound rendering vs the empty-404 flake."""
+
+    def test_empty_body_404_maps_to_retryable(self):
+        """An empty-body 404 is the transient flake: retryable, not not_found."""
+        err = map_twikit_error(NotFound('status: 404, message: ""'))
+        self.assertEqual(err.code, "search_timeline_unavailable")
+        self.assertTrue(err.retryable)
+        self.assertIn("retry with backoff", err.action)
+
+    def test_message_404_stays_non_retryable_not_found(self):
+        """A message-bearing 404 is a real not_found, not the flake."""
+        err = map_twikit_error(NotFound('status: 404, message: "This tweet does not exist"'))
+        self.assertEqual(err.code, "not_found")
+        self.assertFalse(err.retryable)
+
+
 class NewTweetPayloadTests(SimpleTestCase):
     def test_from_tweet(self):
         p = NewTweetPayload.from_tweet(_FakeTweet("123", handle="alice", text="hi"))