From 763c794d2953853459f4af6bf6bb33f4da71726b Mon Sep 17 00:00:00 2001 From: Hayden Garvey <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 06:06:05 +0000 Subject: [PATCH] Re-cut on main: persist processed-mention ledger and improve Grok fallback reply MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-applies the intent of the original fix-spam commit on top of current main, per the review decision to re-cut rather than hand-resolve six months of branch drift. - listener.py: persist replied-to mention IDs (XMCP_PROCESSED_MENTIONS_PATH) and skip them on re-fetch — the inclusive start_time watermark re-returns the boundary mention on every restart. The watermark still advances on skips, the ledger is compacted to XMCP_MAX_PROCESSED_MENTIONS on load, an unreadable ledger degrades to an empty set instead of killing the listener thread, and persistence failures are logged distinctly from reply failures. - agents/team/general.py: replace the "Thinking..." placeholder published when Grok returns nothing with an apology fallback. - env.example: document the new state variables. The original commit's TIMELINE_API_URL port change (8080 -> 8000) is dropped per review feedback — the rest of the repo defaults to 8080. Co-authored-by: linear-code[bot] <222613912+linear-code[bot]@users.noreply.github.com> --- agents/team/general.py | 2 +- env.example | 3 ++ listener.py | 72 +++++++++++++++++++++++++++++++++++++++--- 3 files changed, 72 insertions(+), 5 deletions(-) diff --git a/agents/team/general.py b/agents/team/general.py index e1b6a13..5f7cba7 100644 --- a/agents/team/general.py +++ b/agents/team/general.py @@ -39,7 +39,7 @@ def handle_mention(self, mention: MentionContext) -> AgentReply: f"{wrap_untrusted(mention.text)}" ) if not reply: - reply = "Thinking..." + reply = "Sorry, I'm having trouble processing that. Try again or DM me." card = { "title": f"New mention {mention.mention_id or ''}".strip(), "body": mention.text, diff --git a/env.example b/env.example index e6c6dd1..607bee1 100644 --- a/env.example +++ b/env.example @@ -42,6 +42,9 @@ MCP_DISPATCH_AGENT_ID=mcp-orchestrator # ───────────────────────────────────────────── XMCP_LAST_SEEN_PATH=~/.xmcp/last_seen.txt XMCP_DISPATCH_LAST_SEEN=~/.xmcp/dispatch_last_seen.txt +# Ledger of replied-to mention IDs (suppresses duplicate replies on restart) +XMCP_PROCESSED_MENTIONS_PATH=~/.xmcp/processed_mentions.txt +XMCP_MAX_PROCESSED_MENTIONS=10000 # ───────────────────────────────────────────── # OpenAPI Filtering (optional) diff --git a/listener.py b/listener.py index f9c0801..eabfb61 100644 --- a/listener.py +++ b/listener.py @@ -13,6 +13,14 @@ from agents.registry import register_team, route_mention LAST_SEEN_PATH = Path(os.getenv("XMCP_LAST_SEEN_PATH", "~/.xmcp/last_seen.txt")).expanduser() +PROCESSED_MENTIONS_PATH = Path( + os.getenv("XMCP_PROCESSED_MENTIONS_PATH", "~/.xmcp/processed_mentions.txt") +).expanduser() +# The last-seen watermark has second granularity and start_time is inclusive, +# so the newest processed mention is re-fetched on every restart. The +# processed-mentions ledger exists to suppress that duplicate reply; it only +# needs to cover the replay window, not all history. +MAX_PROCESSED_MENTIONS = int(os.getenv("XMCP_MAX_PROCESSED_MENTIONS", "10000")) POLL_SECONDS = int(os.getenv("POLL_INTERVAL_SECONDS", "60")) PAYMENT_REQUIRED_BACKOFF_SECONDS = int(os.getenv("X_PAYMENT_REQUIRED_BACKOFF_SECONDS", "900")) @@ -34,6 +42,43 @@ def load_last_seen() -> Optional[str]: return LAST_SEEN_PATH.read_text(encoding="utf-8").strip() or None +def load_processed_mentions() -> "set[str]": + """Load recently processed mention IDs, compacting the ledger on the way. + + An unreadable ledger must not kill the listener thread — worst case a + few boundary mentions get a second reply, which is preferable to no + mentions being handled at all. + """ + try: + if not PROCESSED_MENTIONS_PATH.exists(): + return set() + lines = [ + line.strip() + for line in PROCESSED_MENTIONS_PATH.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + except OSError as exc: + print( + f"WARNING: could not read {PROCESSED_MENTIONS_PATH}: {exc}; " + "starting with an empty processed-mentions set (duplicate replies possible)", + flush=True, + ) + return set() + if len(lines) > MAX_PROCESSED_MENTIONS: + lines = lines[-MAX_PROCESSED_MENTIONS:] + try: + PROCESSED_MENTIONS_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") + except OSError as exc: + print(f"WARNING: could not compact {PROCESSED_MENTIONS_PATH}: {exc}", flush=True) + return set(lines) + + +def save_processed_mention(mention_id: str) -> None: + PROCESSED_MENTIONS_PATH.parent.mkdir(parents=True, exist_ok=True) + with PROCESSED_MENTIONS_PATH.open("a", encoding="utf-8") as f: + f.write(f"{mention_id}\n") + + def build_client() -> tweepy.Client: access_token = os.getenv("X_ACCESS_TOKEN") or os.getenv("X_OAUTH_ACCESS_TOKEN") access_secret = os.getenv("X_ACCESS_SECRET") or os.getenv("X_OAUTH_ACCESS_TOKEN_SECRET") @@ -169,6 +214,7 @@ def main() -> None: register_team() last_seen = load_last_seen() + processed_mentions = load_processed_mentions() start_time = datetime.now(timezone.utc) - timedelta(minutes=10) if last_seen: try: @@ -205,10 +251,28 @@ def main() -> None: # the last-seen watermark only ever moves forward. epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) for mention in sorted(mentions.data or [], key=lambda m: m.created_at or epoch): - if not process_mention(client, mention): - # Card push failed: stop here so this mention (and later - # ones) are retried on the next poll. - break + mention_id = str(mention.id) + if mention_id in processed_mentions: + # Already replied (start_time is inclusive, so the boundary + # mention comes back every poll/restart). Still advance the + # watermark so it stops being re-fetched. + print(f"Skipping already processed mention {mention.id}", flush=True) + else: + if not process_mention(client, mention): + # Card push failed: stop here so this mention (and later + # ones) are retried on the next poll. + break + processed_mentions.add(mention_id) + try: + save_processed_mention(mention_id) + except OSError as exc: + # The reply already went out — a persistence failure only + # risks a duplicate after restart, so log it as such + # rather than as a reply failure. + print( + f"WARNING: could not persist processed mention {mention.id}: {exc}", + flush=True, + ) start_time = mention.created_at or datetime.now(timezone.utc) save_last_seen(start_time.isoformat())