feat(sources): X/Twitter social listening (twikit connector) + README update

.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

README.md

@@ -28,16 +28,17 @@
 
 ## 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.
 
 ## Where it listens
 
 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.

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

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,
                 }
             ),
         )

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",
 ]
 

apps/core/sources/connectors/__init__.py

@@ -2,11 +2,13 @@
 from .hackernews import HackerNewsCommentConnector, HackerNewsFeedConnector
 from .reddit import RedditSubRedditConnector
 from .rss import RssConnector
+from .twitter import TwitterSearchConnector
 
 __all__ = [
     "Connector",
     "HackerNewsCommentConnector",
     "HackerNewsFeedConnector",
     "RedditSubRedditConnector",
     "RssConnector",
+    "TwitterSearchConnector",
 ]

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"]

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

@@ -0,0 +1,185 @@
+"""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.
+
+    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,
+        *,
+        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._language = language
+        self._user_agent = user_agent
+        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,
+            )
+
+    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:
+            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")
+        try:
+            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
+            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))